diff --git a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs index 68b41cab5e1..29e43b18c26 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -922,14 +922,6 @@ impl NetworkBeaconProcessor { &metrics::BEACON_BLOB_DELAY_FULL_VERIFICATION, processing_start_time.elapsed().as_millis() as i64, ); - - // If a block is in the da_checker, sync maybe awaiting for an event when block is finally - // imported. A block can become imported both after processing a block or data column. If - // importing a block results in `Imported`, notify. Do not notify of data column errors. - self.send_sync_message(SyncMessage::GossipBlockProcessResult { - block_root, - imported: true, - }); } AvailabilityProcessingStatus::MissingComponents(slot, block_root) => { trace!( @@ -1354,16 +1346,6 @@ impl NetworkBeaconProcessor { // contributing to the partial. } } - - // If a block is in the da_checker, sync maybe awaiting for an event when block is finally - // imported. A block can become imported both after processing a block or data column. If a - // importing a block results in `Imported`, notify. Do not notify of data column errors. - if matches!(result, Ok(AvailabilityProcessingStatus::Imported(_))) { - self.send_sync_message(SyncMessage::GossipBlockProcessResult { - block_root, - imported: true, - }); - } } async fn check_reconstruction_trigger(self: &Arc, slot: Slot, block_root: &Hash256) { @@ -1898,11 +1880,6 @@ impl NetworkBeaconProcessor { if let Err(e) = &result { self.maybe_store_invalid_block(&invalid_block_storage, block_root, &block, e); } - - self.send_sync_message(SyncMessage::GossipBlockProcessResult { - block_root, - imported: matches!(result, Ok(AvailabilityProcessingStatus::Imported(_))), - }); } pub fn process_gossip_voluntary_exit( diff --git a/beacon_node/network/src/network_beacon_processor/sync_methods.rs b/beacon_node/network/src/network_beacon_processor/sync_methods.rs index 8245b5dc0c1..f915574a143 100644 --- a/beacon_node/network/src/network_beacon_processor/sync_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/sync_methods.rs @@ -184,6 +184,9 @@ impl NetworkBeaconProcessor { .await; register_process_result_metrics(&result, metrics::BlockSource::Rpc, "block"); + // Drop the handle to remove the entry from the cache + drop(handle); + // RPC block imported, regardless of process type match result.as_ref() { Ok(AvailabilityProcessingStatus::Imported(hash)) => { @@ -243,9 +246,6 @@ impl NetworkBeaconProcessor { process_type, result: result.into(), }); - - // Drop the handle to remove the entry from the cache - drop(handle); } #[instrument( diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index ad988515328..2bc1ffbdbbf 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -1862,65 +1862,6 @@ async fn test_blobs_by_root_post_fulu_should_return_empty() { assert_eq!(0, actual_count); } -/// Ensure that data column processing that results in block import sends a sync notification -#[tokio::test] -async fn test_data_column_import_notifies_sync() { - if test_spec::().fulu_fork_epoch.is_none() { - return; - } - - let mut rig = TestRig::new(SMALL_CHAIN).await; - let block_root = rig.next_block.canonical_root(); - - // Enqueue the block first to prepare for data column processing - rig.enqueue_gossip_block(); - rig.assert_event_journal_completes(&[WorkType::GossipBlock]) - .await; - rig.receive_sync_messages_with_timeout(Duration::from_millis(100), Some(1)) - .await - .expect("should receive sync message"); - - // Enqueue data columns which should trigger block import when complete - let num_data_columns = rig.next_data_columns.as_ref().map(|c| c.len()).unwrap_or(0); - if num_data_columns > 0 { - for i in 0..num_data_columns { - rig.enqueue_gossip_data_columns(i); - rig.assert_event_journal_completes(&[WorkType::GossipDataColumnSidecar]) - .await; - } - - // Verify block import succeeded - assert_eq!( - rig.head_root(), - block_root, - "block should be imported and become head" - ); - - // Check that sync was notified of the successful import - let sync_messages = rig - .receive_sync_messages_with_timeout(Duration::from_millis(100), Some(1)) - .await - .expect("should receive sync message"); - - // Verify we received the expected GossipBlockProcessResult message - assert_eq!( - sync_messages.len(), - 1, - "should receive exactly one sync message" - ); - match &sync_messages[0] { - SyncMessage::GossipBlockProcessResult { - block_root: msg_block_root, - imported, - } => { - assert_eq!(*msg_block_root, block_root, "block root should match"); - assert!(*imported, "block should be marked as imported"); - } - other => panic!("expected GossipBlockProcessResult, got {:?}", other), - } - } -} - #[tokio::test] async fn test_data_columns_by_range_request_only_returns_requested_columns() { if test_spec::().fulu_fork_epoch.is_none() { diff --git a/beacon_node/network/src/sync/block_lookups/mod.rs b/beacon_node/network/src/sync/block_lookups/mod.rs index a265373e3fc..0cbeb5ee4ee 100644 --- a/beacon_node/network/src/sync/block_lookups/mod.rs +++ b/beacon_node/network/src/sync/block_lookups/mod.rs @@ -482,39 +482,6 @@ impl BlockLookups { self.on_lookup_result(lookup_id, lookup_result, "processing_result", cx); } - pub fn on_external_processing_result( - &mut self, - block_root: Hash256, - imported: bool, - cx: &mut SyncNetworkContext, - ) { - let Some((id, lookup)) = self - .single_block_lookups - .iter_mut() - .find(|(_, lookup)| lookup.is_for_block(block_root)) - else { - // Ok to ignore gossip process events - return; - }; - - let lookup_result = if imported { - Ok(LookupResult::Completed) - } else { - // A lookup may be in the following state: - // - Block awaiting processing from a different source - // - Blobs downloaded processed, and inserted into the da_checker - // - // At this point the block fails processing (e.g. execution engine offline) and it is - // removed from the da_checker. Note that ALL components are removed from the da_checker - // so when we re-download and process the block we get the error - // MissingComponentsAfterAllProcessed and get stuck. - lookup.reset_requests(); - lookup.continue_requests(cx) - }; - let id = *id; - self.on_lookup_result(id, lookup_result, "external_processing_result", cx); - } - /// Makes progress on the immediate children of `block_root` pub fn continue_child_lookups(&mut self, block_root: Hash256, cx: &mut SyncNetworkContext) { let mut lookup_results = vec![]; // < need to buffer lookup results to not re-borrow &mut self 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 8eb58da4e6e..209878afd6e 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 @@ -3,11 +3,10 @@ use crate::network_beacon_processor::BlockProcessingResult; use crate::sync::block_lookups::{BlockDownloadResponse, CustodyDownloadResponse}; use crate::sync::manager::BlockProcessType; use crate::sync::network_context::{ - LookupRequestResult, PeerGroup, ReqId, RpcRequestSendError, RpcResponseError, - SendErrorProcessor, SyncNetworkContext, + PeerGroup, ReqId, RpcRequestSendError, RpcResponseError, SendErrorProcessor, SyncNetworkContext, }; -use beacon_chain::BeaconChainTypes; use beacon_chain::block_verification_types::AsBlock; +use beacon_chain::{BeaconChainTypes, BlockProcessStatus}; use educe::Educe; use lighthouse_network::service::api_types::Id; use parking_lot::RwLock; @@ -139,12 +138,6 @@ impl SingleBlockLookup { } } - /// Reset the status of all requests (used on block processing failure) - pub fn reset_requests(&mut self) { - self.block_request = BlockRequest::new(); - self.data_request = DataRequest::WaitingForBlock; - } - /// Return the slot of this lookup's block if it's currently cached pub fn peek_downloaded_block_slot(&self) -> Option { self.block_request @@ -219,7 +212,7 @@ impl SingleBlockLookup { // === Block request === self.block_request.state.maybe_start_downloading(|| { - cx.block_lookup_request(self.id, self.peers.clone(), self.block_root) + send_block_request(self.id, self.peers.clone(), self.block_root, cx) })?; if self.awaiting_parent.is_none() && let Some(data) = self.block_request.state.maybe_start_processing() @@ -252,11 +245,12 @@ impl SingleBlockLookup { } DataRequest::Request { slot, state } => { state.maybe_start_downloading(|| { - cx.custody_lookup_request( + send_custody_request( self.id, + self.peers.clone(), self.block_root, *slot, - self.peers.clone(), + cx, ) })?; // Wait for the parent to be imported, data column processing result handle does @@ -401,6 +395,137 @@ impl SingleBlockLookup { } } +fn send_block_request( + lookup_id: Id, + lookup_peers: PeerSet, + block_root: Hash256, + cx: &mut SyncNetworkContext, +) -> Result>>, RpcRequestSendError> { + let active_request_count_by_peer = cx.active_request_count_by_peer(); + let Some(peer_id) = lookup_peers + .read() + .iter() + .map(|peer| { + ( + // Prefer peers with less overall requests + active_request_count_by_peer.get(peer).copied().unwrap_or(0), + // Random factor to break ties, otherwise the PeerID breaks ties + rand::random::(), + peer, + ) + }) + .min() + .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 + // attestations for said block. + // Lookup sync event safety: If a lookup requires peers to make progress, and does + // not receive any new peers for some time it will be dropped. If it receives a new + // peer it must attempt to make progress. + return Ok(LookupRequestResult::RequestNotSent("no peers")); + }; + + match cx.chain.get_block_process_status(&block_root) { + // Unknown block, continue request to download + BlockProcessStatus::Unknown => {} + // Block is known but processing. The block may turn out to be invalid, so we want sync to + // NOT mark the request as complete yet. The ideal flow would be: + // - Wait for processing to complete + // - Only if there is an error re-download and re-process + // But implementing this introduces complexity and the risk for the lookup to get stuck. + // Instead we always re-download the block eagerly and de-duplicate the processing. So in + // the happy case we just download the block again if the lookup is created while execution + // processing the block. + BlockProcessStatus::NotValidated(..) => {} + // Block is fully validated. If it's not yet imported it's waiting for missing block + // components. Consider this request completed and do nothing. + BlockProcessStatus::ExecutionValidated(block) => { + return Ok(LookupRequestResult::MarkRequestAsComplete( + "block execution validated", + block, + )); + } + }; + + let id = cx.block_lookup_request(lookup_id, peer_id, block_root)?; + Ok(LookupRequestResult::RequestSent(id)) +} + +fn send_custody_request( + lookup_id: Id, + lookup_peers: PeerSet, + block_root: Hash256, + block_slot: Slot, + cx: &mut SyncNetworkContext, +) -> Result>, RpcRequestSendError> { + let custody_indexes_imported = cx + .chain + .cached_data_column_indexes(&block_root, block_slot) + .unwrap_or_default(); + + // Include only the blob indexes not yet imported (received through gossip) + let mut custody_indexes_to_fetch = cx + .chain + .sampling_columns_for_epoch(block_slot.epoch(T::EthSpec::slots_per_epoch())) + .iter() + .copied() + .filter(|index| !custody_indexes_imported.contains(index)) + .collect::>(); + custody_indexes_to_fetch.sort_unstable(); + + if custody_indexes_to_fetch.is_empty() { + // No indexes required, do not issue any request + return Ok(LookupRequestResult::MarkRequestAsComplete( + "no indices to fetch", + vec![], + )); + } + let id = cx.custody_lookup_request( + lookup_id, + block_root, + &custody_indexes_to_fetch, + lookup_peers, + )?; + Ok(LookupRequestResult::RequestSent(id)) +} + +#[allow(dead_code)] +fn send_payload_request( + lookup_id: Id, + lookup_peers: PeerSet, + block_root: Hash256, + cx: &mut SyncNetworkContext, +) -> Result, RpcRequestSendError> { + // Skip the download if fork-choice already saw this envelope (e.g. imported via gossip + // before the lookup got here). + if cx.chain.envelope_is_known_to_fork_choice(&block_root) { + return Ok(LookupRequestResult::MarkRequestAsComplete( + "envelope already known to fork-choice", + (), + )); + } + + let active_request_count_by_peer = cx.active_request_count_by_peer(); + let Some(peer_id) = lookup_peers + .read() + .iter() + .map(|peer| { + ( + active_request_count_by_peer.get(peer).copied().unwrap_or(0), + rand::random::(), + peer, + ) + }) + .min() + .map(|(_, _, peer)| *peer) + else { + return Ok(LookupRequestResult::RequestNotSent("no peers")); + }; + let id = cx.payload_lookup_request(lookup_id, peer_id, block_root)?; + Ok(LookupRequestResult::RequestSent(id)) +} + #[derive(Debug, Clone)] pub struct DownloadResult { pub value: T, @@ -418,6 +543,20 @@ impl DownloadResult { } } +pub enum LookupRequestResult { + /// A request is sent. Sync MUST receive an event from the network in the future for either: + /// completed response or failed request + RequestSent(I), + /// No request is sent, and no further action is necessary to consider this request completed. + /// Includes a reason why this request is not needed. + MarkRequestAsComplete(&'static str, T), + /// No request is sent, but the request is not completed. Sync MUST receive some future event + /// that makes progress on the request. For example: request is processing from a different + /// source (i.e. block received from gossip) and sync MUST receive an event with that processing + /// result. + RequestNotSent(&'static str), +} + #[derive(IntoStaticStr)] pub enum State { AwaitingDownload(/* reason */ &'static str), @@ -505,10 +644,10 @@ impl SingleLookupRequestState { if self.is_awaiting_download() { match request_fn().map_err(LookupRequestError::SendFailedNetwork)? { LookupRequestResult::RequestSent(req_id) => self.on_download_start(req_id)?, - LookupRequestResult::NoRequestNeeded(reason, value) => { + LookupRequestResult::MarkRequestAsComplete(reason, value) => { self.on_completed_request(reason, value)? } - LookupRequestResult::Pending(reason) => { + LookupRequestResult::RequestNotSent(reason) => { self.update_awaiting_download_status(reason) } } diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index 166c65b6e1a..66bb13ae988 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -182,11 +182,6 @@ pub enum SyncMessage { process_type: BlockProcessType, result: BlockProcessingResult, }, - - /// A gossip-received component has completed processing and the block may now be imported. - /// In Fulu this is sent after block or blob processing. In Gloas this is also sent after - /// data column or payload envelope processing triggers availability. - GossipBlockProcessResult { block_root: Hash256, imported: bool }, } /// The type of processing specified for a received block. @@ -907,14 +902,6 @@ impl SyncManager { } => self .block_lookups .on_processing_result(process_type, result, &mut self.network), - SyncMessage::GossipBlockProcessResult { - block_root, - imported, - } => self.block_lookups.on_external_processing_result( - block_root, - imported, - &mut self.network, - ), SyncMessage::BatchProcessed { sync_type, result } => match sync_type { ChainSegmentProcessId::RangeBatchId(chain_id, epoch) => { self.range_sync.handle_block_process_result( diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 1e35c0a72f6..d8775b2bfb6 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -21,7 +21,7 @@ use crate::sync::block_sidecar_coupling::CouplingError; use crate::sync::range_data_column_batch_request::RangeDataColumnBatchRequest; use beacon_chain::block_verification_types::LookupBlock; use beacon_chain::block_verification_types::{AsBlock, RangeSyncBlock}; -use beacon_chain::{BeaconChain, BeaconChainTypes, BlockProcessStatus, EngineState}; +use beacon_chain::{BeaconChain, BeaconChainTypes, EngineState}; use custody::CustodyRequestResult; use fnv::FnvHashMap; use lighthouse_network::rpc::methods::{BlobsByRangeRequest, DataColumnsByRangeRequest}; @@ -53,8 +53,8 @@ use task_executor::TaskExecutor; use tokio::sync::mpsc; use tracing::{Span, debug, debug_span, error, warn}; use types::{ - BlobSidecar, BlockImportSource, ColumnIndex, DataColumnSidecar, DataColumnSidecarList, EthSpec, - ForkContext, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, + BlobSidecar, ColumnIndex, DataColumnSidecar, DataColumnSidecarList, EthSpec, ForkContext, + Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, }; pub mod custody; @@ -176,20 +176,6 @@ impl PeerGroup { /// Sequential ID that uniquely identifies ReqResp outgoing requests pub type ReqId = u32; -pub enum LookupRequestResult { - /// A request is sent. Sync MUST receive an event from the network in the future for either: - /// completed response or failed request - RequestSent(I), - /// No request is sent, and no further action is necessary to consider this request completed. - /// Includes a reason why this request is not needed. - NoRequestNeeded(&'static str, T), - /// No request is sent, but the request is not completed. Sync MUST receive some future event - /// that makes progress on the request. For example: request is processing from a different - /// source (i.e. block received from gossip) and sync MUST receive an event with that processing - /// result. - Pending(&'static str), -} - /// Wraps a Network channel to employ various RPC related network functionality for the Sync manager. This includes management of a global RPC request Id. pub struct SyncNetworkContext { /// The network channel to relay messages to the Network service. @@ -416,7 +402,7 @@ impl SyncNetworkContext { } } - fn active_request_count_by_peer(&self) -> HashMap { + pub fn active_request_count_by_peer(&self) -> HashMap { let Self { network_send: _, request_id: _, @@ -818,67 +804,9 @@ impl SyncNetworkContext { pub fn block_lookup_request( &mut self, lookup_id: SingleLookupId, - lookup_peers: Arc>>, + peer_id: PeerId, block_root: Hash256, - ) -> Result>>, RpcRequestSendError> { - let active_request_count_by_peer = self.active_request_count_by_peer(); - let Some(peer_id) = lookup_peers - .read() - .iter() - .map(|peer| { - ( - // Prefer peers with less overall requests - active_request_count_by_peer.get(peer).copied().unwrap_or(0), - // Random factor to break ties, otherwise the PeerID breaks ties - rand::random::(), - peer, - ) - }) - .min() - .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 - // attestations for said block. - // Lookup sync event safety: If a lookup requires peers to make progress, and does - // not receive any new peers for some time it will be dropped. If it receives a new - // peer it must attempt to make progress. - return Ok(LookupRequestResult::Pending("no peers")); - }; - - match self.chain.get_block_process_status(&block_root) { - // Unknown block, continue request to download - BlockProcessStatus::Unknown => {} - // Block is known and currently processing. Imports from gossip and HTTP API insert the - // block in the da_cache. However, HTTP API is unable to notify sync when it completes - // block import. Returning `Pending` here will result in stuck lookups if the block is - // importing from sync. - BlockProcessStatus::NotValidated(_, source) => match source { - BlockImportSource::Gossip => { - // Lookup sync event safety: If the block is currently in the processing cache, we - // are guaranteed to receive a `SyncMessage::GossipBlockProcessResult` that will - // make progress on this lookup - return Ok(LookupRequestResult::Pending("block in processing cache")); - } - BlockImportSource::Lookup - | BlockImportSource::RangeSync - | BlockImportSource::HttpApi => { - // Lookup, RangeSync or HttpApi block import don't emit the GossipBlockProcessResult - // event. If a lookup happens to be created during block import from one of - // those sources just import the block twice. Otherwise the lookup will get - // stuck. Double imports are fine, they just waste resources. - } - }, - // Block is fully validated. If it's not yet imported it's waiting for missing block - // components. Consider this request completed and do nothing. - BlockProcessStatus::ExecutionValidated(block) => { - return Ok(LookupRequestResult::NoRequestNeeded( - "block execution validated", - block, - )); - } - } - + ) -> Result { let id = SingleLookupReqId { lookup_id, req_id: self.next_id(), @@ -928,43 +856,16 @@ impl SyncNetworkContext { request_span, ); - Ok(LookupRequestResult::RequestSent(id.req_id)) + Ok(id.req_id) } /// Request a payload envelope for a block root via PayloadEnvelopesByRoot RPC. - #[allow(dead_code)] pub fn payload_lookup_request( &mut self, lookup_id: SingleLookupId, - lookup_peers: Arc>>, + peer_id: PeerId, block_root: Hash256, - ) -> Result, RpcRequestSendError> { - // Skip the download if fork-choice already saw this envelope (e.g. imported via gossip - // before the lookup got here). - if self.chain.envelope_is_known_to_fork_choice(&block_root) { - return Ok(LookupRequestResult::NoRequestNeeded( - "envelope already known to fork-choice", - (), - )); - } - - let active_request_count_by_peer = self.active_request_count_by_peer(); - let Some(peer_id) = lookup_peers - .read() - .iter() - .map(|peer| { - ( - active_request_count_by_peer.get(peer).copied().unwrap_or(0), - rand::random::(), - peer, - ) - }) - .min() - .map(|(_, _, peer)| *peer) - else { - return Ok(LookupRequestResult::Pending("no peers")); - }; - + ) -> Result { let id = SingleLookupReqId { lookup_id, req_id: self.next_id(), @@ -1004,7 +905,7 @@ impl SyncNetworkContext { Span::none(), ); - Ok(LookupRequestResult::RequestSent(id.req_id)) + Ok(id.req_id) } /// Request to send a single `data_columns_by_root` request to the network. pub fn data_column_lookup_request( @@ -1013,7 +914,7 @@ impl SyncNetworkContext { peer_id: PeerId, request: DataColumnsByRootSingleBlockRequest, expect_max_responses: bool, - ) -> Result, &'static str> { + ) -> Result { let id = DataColumnsByRootRequestId { id: self.next_id(), requester, @@ -1049,7 +950,7 @@ impl SyncNetworkContext { Span::none(), ); - Ok(LookupRequestResult::RequestSent(id)) + Ok(id) } /// Request to fetch all needed custody columns of a specific block. This function may not send @@ -1060,32 +961,9 @@ impl SyncNetworkContext { &mut self, lookup_id: SingleLookupId, block_root: Hash256, - block_slot: Slot, + column_indices: &[ColumnIndex], lookup_peers: Arc>>, - ) -> Result>, RpcRequestSendError> { - let custody_indexes_imported = self - .chain - .cached_data_column_indexes(&block_root, block_slot) - .unwrap_or_default(); - - // Include only the blob indexes not yet imported (received through gossip) - let mut custody_indexes_to_fetch = self - .chain - .sampling_columns_for_epoch(block_slot.epoch(T::EthSpec::slots_per_epoch())) - .iter() - .copied() - .filter(|index| !custody_indexes_imported.contains(index)) - .collect::>(); - custody_indexes_to_fetch.sort_unstable(); - - if custody_indexes_to_fetch.is_empty() { - // No indexes required, do not issue any request - return Ok(LookupRequestResult::NoRequestNeeded( - "no indices to fetch", - vec![], - )); - } - + ) -> Result { let id = SingleLookupReqId { lookup_id, req_id: self.next_id(), @@ -1093,7 +971,7 @@ impl SyncNetworkContext { debug!( ?block_root, - indices = ?custody_indexes_to_fetch, + indices = ?column_indices, %id, "Starting custody columns request" ); @@ -1102,7 +980,7 @@ impl SyncNetworkContext { let mut request = ActiveCustodyRequest::new( block_root, CustodyId { requester }, - &custody_indexes_to_fetch, + &column_indices, lookup_peers, ); @@ -1113,7 +991,7 @@ impl SyncNetworkContext { // created cannot return data immediately, it must send some request to the network // first. And there must exist some request, `custody_indexes_to_fetch` is not empty. self.custody_by_root_requests.insert(requester, request); - Ok(LookupRequestResult::RequestSent(id.req_id)) + Ok(id.req_id) } Err(e) => Err(match e { CustodyRequestError::NoPeer(column_index) => { diff --git a/beacon_node/network/src/sync/network_context/custody.rs b/beacon_node/network/src/sync/network_context/custody.rs index e74b74ec08e..f1ea745fbe3 100644 --- a/beacon_node/network/src/sync/network_context/custody.rs +++ b/beacon_node/network/src/sync/network_context/custody.rs @@ -16,7 +16,7 @@ use tracing::{Span, debug, debug_span, warn}; use types::{DataColumnSidecar, Hash256, data::ColumnIndex}; use types::{DataColumnSidecarList, EthSpec}; -use super::{LookupRequestResult, PeerGroup, RpcResponseResult, SyncNetworkContext}; +use super::{PeerGroup, RpcResponseResult, SyncNetworkContext}; const MAX_STALE_NO_PEERS_DURATION: Duration = Duration::from_secs(30); @@ -297,7 +297,7 @@ impl ActiveCustodyRequest { } for (peer_id, indices) in columns_to_request_by_peer.into_iter() { - let request_result = cx + let req_id = cx .data_column_lookup_request( DataColumnsByRootRequester::Custody(self.custody_id), peer_id, @@ -318,38 +318,32 @@ impl ActiveCustodyRequest { ) .map_err(Error::SendFailed)?; - match request_result { - LookupRequestResult::RequestSent(req_id) => { - *self.peer_attempts.entry(peer_id).or_insert(0) += 1; + *self.peer_attempts.entry(peer_id).or_insert(0) += 1; - let client = cx.network_globals().client(&peer_id).kind; - let batch_columns_req_span = debug_span!( - "batch_columns_req", - %peer_id, - %client, - ); - let _guard = batch_columns_req_span.clone().entered(); - for column_index in &indices { - let column_request = self - .column_requests - .get_mut(column_index) - // Should never happen: column_index is iterated from column_requests - .ok_or(Error::BadState("unknown column_index".to_owned()))?; - - column_request.on_download_start(req_id)?; - } - - self.active_batch_columns_requests.insert( - req_id, - ActiveBatchColumnsRequest { - indices, - span: batch_columns_req_span, - }, - ); - } - LookupRequestResult::NoRequestNeeded(..) => unreachable!(), - LookupRequestResult::Pending(_) => unreachable!(), + let client = cx.network_globals().client(&peer_id).kind; + let batch_columns_req_span = debug_span!( + "batch_columns_req", + %peer_id, + %client, + ); + let _guard = batch_columns_req_span.clone().entered(); + for column_index in &indices { + let column_request = self + .column_requests + .get_mut(column_index) + // Should never happen: column_index is iterated from column_requests + .ok_or(Error::BadState("unknown column_index".to_owned()))?; + + column_request.on_download_start(req_id)?; } + + self.active_batch_columns_requests.insert( + req_id, + ActiveBatchColumnsRequest { + indices, + span: batch_columns_req_span, + }, + ); } Ok(None) diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs index 5642f7846a6..91227d77f85 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -1235,12 +1235,6 @@ impl TestRig { self.assert_empty_network(); } - fn assert_pending_lookup_sync(&self) { - assert!(self.created_lookups() > 0, "no created lookups"); - assert_eq!(self.dropped_lookups(), 0, "some dropped lookups"); - assert_eq!(self.completed_lookups(), 0, "some completed lookups"); - } - /// Assert there is at least one range sync chain created and that all sync chains completed pub(super) fn assert_successful_range_sync(&self) { assert!( @@ -1330,15 +1324,6 @@ impl TestRig { genesis_fork().fulu_enabled().then(Self::default) } - fn new_after_deneb_before_fulu() -> Option { - let fork = genesis_fork(); - if fork.deneb_enabled() && !fork.fulu_enabled() { - Some(Self::default()) - } else { - None - } - } - pub fn new_fulu_peer_test(fulu_test_type: FuluTestType) -> Option { genesis_fork().fulu_enabled().then(|| { Self::new(TestRigConfig { @@ -1673,56 +1658,6 @@ impl TestRig { } } - fn insert_block_to_da_checker_as_pre_execution(&mut self, block: Arc>) { - self.log(&format!( - "Inserting block to availability_cache as pre_execution_block {:?}", - block.canonical_root() - )); - self.harness - .chain - .data_availability_checker - .put_pre_execution_block(block.canonical_root(), block, BlockImportSource::Gossip) - .unwrap(); - } - - fn simulate_block_gossip_processing_becomes_invalid(&mut self, block_root: Hash256) { - self.log(&format!( - "Marking block {block_root:?} in da_checker as execution error" - )); - self.harness - .chain - .data_availability_checker - .remove_block_on_execution_error(&block_root); - - self.send_sync_message(SyncMessage::GossipBlockProcessResult { - block_root, - imported: false, - }); - } - - async fn simulate_block_gossip_processing_becomes_valid( - &mut self, - block: Arc>, - ) { - let block_root = block.canonical_root(); - - match self.import_block_to_da_checker(block).await { - AvailabilityProcessingStatus::Imported(block_root) => { - self.log(&format!( - "insert block to da_checker and it imported {block_root:?}" - )); - } - AvailabilityProcessingStatus::MissingComponents(_, _) => { - panic!("block not imported after adding to da_checker"); - } - } - - self.send_sync_message(SyncMessage::GossipBlockProcessResult { - block_root, - imported: false, - }); - } - fn requests_count(&self) -> HashMap<&'static str, usize> { let mut requests_count = HashMap::new(); for (request, _) in &self.requests { @@ -2294,48 +2229,6 @@ async fn block_in_da_checker_skips_download() { ); } -#[tokio::test] -async fn block_in_processing_cache_becomes_invalid() { - let Some(mut r) = TestRig::new_after_deneb_before_fulu() else { - return; - }; - r.build_chain(1).await; - let block = r.block_at_slot(1); - r.insert_block_to_da_checker_as_pre_execution(block.clone()); - r.trigger_with_last_block(); - r.simulate(SimulateConfig::happy_path()).await; - r.assert_pending_lookup_sync(); - // Here the only active lookup is waiting for the block to finish processing - - // Simulate invalid block, removing it from processing cache - r.simulate_block_gossip_processing_becomes_invalid(block.canonical_root()); - // Should download block, then issue blobs request - r.simulate(SimulateConfig::happy_path()).await; - r.assert_successful_lookup_sync(); -} - -#[tokio::test] -async fn block_in_processing_cache_becomes_valid_imported() { - let Some(mut r) = TestRig::new_after_deneb_before_fulu() else { - return; - }; - r.build_chain(1).await; - let block = r.block_at_slot(1); - r.insert_block_to_da_checker_as_pre_execution(block.clone()); - r.trigger_with_last_block(); - r.simulate(SimulateConfig::happy_path()).await; - r.assert_pending_lookup_sync(); - // Here the only active lookup is waiting for the block to finish processing - - // Resolve the block from processing step - r.simulate_block_gossip_processing_becomes_valid(block) - .await; - // Should not trigger block or blob request - r.assert_empty_network(); - // Resolve blob and expect lookup completed - r.assert_no_active_lookups(); -} - macro_rules! fulu_peer_matrix_tests { ( [$($name:ident => $variant:expr),+ $(,)?]