Skip to content

Clean up network_context from da_checker oracles #80

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
dapplion wants to merge 1 commit into
base: unstable
Choose a base branch
from
Open

Clean up network_context from da_checker oracles #80

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 0 additions & 23 deletions beacon_node/network/src/network_beacon_processor/gossip_methods.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -922,14 +922,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
&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!(
Expand Down Expand Up @@ -1354,16 +1346,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
// 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<Self>, slot: Slot, block_root: &Hash256) {
Expand Down Expand Up @@ -1898,11 +1880,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
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(
Expand Down
6 changes: 3 additions & 3 deletions beacon_node/network/src/network_beacon_processor/sync_methods.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -184,6 +184,9 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
.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)) => {
Expand Down Expand Up @@ -243,9 +246,6 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
process_type,
result: result.into(),
});

// Drop the handle to remove the entry from the cache
drop(handle);
}

#[instrument(
Expand Down
59 changes: 0 additions & 59 deletions beacon_node/network/src/network_beacon_processor/tests.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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::<E>().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::<E>().fulu_fork_epoch.is_none() {
Expand Down
33 changes: 0 additions & 33 deletions beacon_node/network/src/sync/block_lookups/mod.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -482,39 +482,6 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
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<T>,
) {
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<T>) {
let mut lookup_results = vec![]; // < need to buffer lookup results to not re-borrow &mut self
Expand Down
167 changes: 153 additions & 14 deletions beacon_node/network/src/sync/block_lookups/single_block_lookup.rs
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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;
Expand Down Expand Up @@ -139,12 +138,6 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
}
}

/// 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<Slot> {
self.block_request
Expand Down Expand Up @@ -219,7 +212,7 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {

// === 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()
Expand Down Expand Up @@ -252,11 +245,12 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
}
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
Expand Down Expand Up @@ -401,6 +395,137 @@ impl<T: BeaconChainTypes> SingleBlockLookup<T> {
}
}

fn send_block_request<T: BeaconChainTypes>(
lookup_id: Id,
lookup_peers: PeerSet,
block_root: Hash256,
cx: &mut SyncNetworkContext<T>,
) -> Result<LookupRequestResult<Arc<SignedBeaconBlock<T::EthSpec>>>, 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::<u32>(),
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<T: BeaconChainTypes>(
lookup_id: Id,
lookup_peers: PeerSet,
block_root: Hash256,
block_slot: Slot,
cx: &mut SyncNetworkContext<T>,
) -> Result<LookupRequestResult<DataColumnSidecarList<T::EthSpec>>, 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::<Vec<_>>();
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<T: BeaconChainTypes>(
lookup_id: Id,
lookup_peers: PeerSet,
block_root: Hash256,
cx: &mut SyncNetworkContext<T>,
) -> Result<LookupRequestResult<()>, 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::<u32>(),
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<T: Clone> {
pub value: T,
Expand All @@ -418,6 +543,20 @@ impl<T: Clone> DownloadResult<T> {
}
}

pub enum LookupRequestResult<T, I = ReqId> {
/// 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<T: Clone> {
AwaitingDownload(/* reason */ &'static str),
Expand Down Expand Up @@ -505,10 +644,10 @@ impl<T: Clone> SingleLookupRequestState<T> {
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)
}
}
Expand Down
Loading
Loading