gnosis/lighthouse
1
0
Fork 0
mirror of https://github.com/sigp/lighthouse.git synced 2026-06-15 17:58:23 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge remote-tracking branch 'sigp/unstable' into gloas-lookup-sync-fixes

Browse Source
This commit is contained in:
dapplion
2026-06-06 11:32:38 +02:00
parent 6be84997f1 8e4df4abab
commit 7d71c47a66
42 changed files with 1317 additions and 845 deletions
Download Patch File Download Diff File Expand all files Collapse all files

63
beacon_node/network/src/network_beacon_processor/gossip_methods.rs
View File

@@ -171,17 +171,6 @@ impl<E: EthSpec> FailedAtt<E> {
}
}
/// `MessageAcceptance` doesn't implement clone so we do a manual match here.
/// TODO: remove this once `Clone` is available on this type:
/// https://github.com/libp2p/rust-libp2p/pull/6445
fn clone_message_acceptance(a: &MessageAcceptance) -> MessageAcceptance {
match a {
MessageAcceptance::Accept => MessageAcceptance::Accept,
MessageAcceptance::Reject => MessageAcceptance::Reject,
MessageAcceptance::Ignore => MessageAcceptance::Ignore,
}
}
impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
/* Auxiliary functions */
@@ -933,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!(
@@ -1365,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) {
@@ -1654,9 +1625,14 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
crit!(error = %e, "Internal block gossip validation error. Availability check during gossip validation");
return None;
}
Err(e @ BlockError::InternalError(_))
| Err(e @ BlockError::EnvelopeBlockRootUnknown(_))
| Err(e @ BlockError::OptimisticSyncNotSupported { .. }) => {
// This error variant cannot be reached when doing gossip block validation: a block has
// no envelope to verify, and `BlockError::EnvelopeError` is only ever produced by the
// envelope import pipeline.
Err(e @ BlockError::EnvelopeError(_)) => {
crit!(error = %e, "Internal block gossip validation error. Envelope error during gossip validation");
return None;
}
Err(e @ BlockError::InternalError(_)) => {
error!(error = %e, "Internal block gossip validation error");
return None;
}
@@ -1904,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(
@@ -2013,11 +1984,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
}
};
self.propagate_validation_result(
message_id,
peer_id,
clone_message_acceptance(&validation_result),
);
self.propagate_validation_result(message_id, peer_id, validation_result);
if let Some(slashing) = verified_slashing_opt {
metrics::inc_counter(&metrics::BEACON_PROCESSOR_PROPOSER_SLASHING_VERIFIED_TOTAL);
@@ -2079,11 +2046,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
}
};
self.propagate_validation_result(
message_id,
peer_id,
clone_message_acceptance(&validation_result),
);
self.propagate_validation_result(message_id, peer_id, validation_result);
if let Some(slashing) = verified_slashing_opt {
metrics::inc_counter(&metrics::BEACON_PROCESSOR_ATTESTER_SLASHING_VERIFIED_TOTAL);
@@ -3748,7 +3711,11 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
EnvelopeError::PriorToFinalization { .. }
| EnvelopeError::BeaconChainError(_)
| EnvelopeError::BeaconStateError(_)
| EnvelopeError::ImportError(_) => {
// The following variants are produced during envelope import, not gossip
// verification, so they cannot be reached here. Ignore them to be safe.
| EnvelopeError::OptimisticSyncNotSupported { .. }
| EnvelopeError::BlockRootNotInForkChoice(_)
| EnvelopeError::InternalError(_) => {
self.propagate_validation_result(
message_id,
peer_id,

8
beacon_node/network/src/network_beacon_processor/sync_methods.rs
View File

@@ -367,7 +367,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
)
.await
}
Err(e) => Err(e),
Err(e) => Err(e.into()),
};
// TODO(gloas): structured penalty classification arrives with the envelope lookup state
@@ -1030,6 +1030,10 @@ impl From<Result<AvailabilityProcessingStatus, BlockError>> for BlockProcessingR
None
}
}
BlockError::EnvelopeError(_) => {
// TODO(gloas): penalize correctly in range sync PR
None
}
// Remaining invalid blocks: penalize the block peer. Listed explicitly so a
// new `BlockError` variant forces a compile error here.
BlockError::FutureSlot { .. }
@@ -1049,8 +1053,6 @@ impl From<Result<AvailabilityProcessingStatus, BlockError>> for BlockProcessingR
| BlockError::ParentExecutionPayloadInvalid { .. }
| BlockError::KnownInvalidExecutionPayload(_)
| BlockError::Slashable
| BlockError::EnvelopeBlockRootUnknown(_)
| BlockError::OptimisticSyncNotSupported { .. }
| BlockError::InvalidBlobCount { .. }
| BlockError::BidParentRootMismatch { .. } => block_peer_penalty(&e),
};

104
beacon_node/network/src/network_beacon_processor/tests.rs
View File

@@ -6,7 +6,7 @@ use crate::{
ChainSegmentProcessId, DuplicateCache, InvalidBlockStorage, NetworkBeaconProcessor,
},
service::NetworkMessage,
sync::{SyncMessage, manager::BlockProcessType},
sync::manager::BlockProcessType,
};
use beacon_chain::block_verification_types::LookupBlock;
use beacon_chain::custody_context::NodeCustodyType;
@@ -76,7 +76,6 @@ struct TestRig {
beacon_processor_tx: BeaconProcessorSend<E>,
work_journal_rx: mpsc::Receiver<&'static str>,
network_rx: mpsc::UnboundedReceiver<NetworkMessage<E>>,
sync_rx: mpsc::UnboundedReceiver<SyncMessage<E>>,
duplicate_cache: DuplicateCache,
network_beacon_processor: Arc<NetworkBeaconProcessor<T>>,
_harness: BeaconChainHarness<T>,
@@ -270,7 +269,7 @@ impl TestRig {
beacon_processor_rx,
} = BeaconProcessorChannels::new(&beacon_processor_config);
let (sync_tx, sync_rx) = mpsc::unbounded_channel();
let (sync_tx, _sync_rx) = mpsc::unbounded_channel();
// Default metadata
let meta_data = if spec.is_peer_das_scheduled() {
@@ -375,7 +374,6 @@ impl TestRig {
beacon_processor_tx,
work_journal_rx,
network_rx,
sync_rx,
duplicate_cache,
network_beacon_processor,
_harness: harness,
@@ -844,45 +842,6 @@ impl TestRig {
Some(events)
}
}
/// Listen for sync messages and collect them for a specified duration or until reaching a count.
///
/// Returns None if no messages were received, or Some(Vec) containing the received messages.
pub async fn receive_sync_messages_with_timeout(
&mut self,
timeout: Duration,
count: Option<usize>,
) -> Option<Vec<SyncMessage<E>>> {
let mut events = vec![];
let timeout_future = tokio::time::sleep(timeout);
tokio::pin!(timeout_future);
loop {
// Break if we've received the requested count of messages
if let Some(target_count) = count
&& events.len() >= target_count
{
break;
}
tokio::select! {
_ = &mut timeout_future => break,
maybe_msg = self.sync_rx.recv() => {
match maybe_msg {
Some(msg) => events.push(msg),
None => break, // Channel closed
}
}
}
}
if events.is_empty() {
None
} else {
Some(events)
}
}
}
fn junk_peer_id() -> PeerId {
@@ -1862,65 +1821,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() {

5
beacon_node/network/src/service.rs
View File

@@ -883,10 +883,7 @@ impl<T: BeaconChainTypes> NetworkService<T> {
fork_context.update_current_fork(*new_fork_name, new_fork_digest, current_epoch);
if self.beacon_chain.spec.is_peer_das_scheduled() {
let next_fork_digest = fork_context
.next_fork_digest()
.unwrap_or_else(|| fork_context.current_fork_digest());
self.libp2p.update_nfd(next_fork_digest);
self.libp2p.update_nfd(fork_context.next_fork_digest());
}
self.libp2p.update_fork_version(new_enr_fork_id);

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

@@ -587,68 +587,6 @@ impl<T: BeaconChainTypes> BlockLookups<T> {
self.on_lookup_result(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) = self
.single_block_lookups
.iter()
.find(|(_, lookup)| lookup.is_for_block(block_root))
.map(|(id, _)| *id)
else {
// Ok to ignore gossip process events
return;
};
if imported {
// The block is imported into fork choice. Unblock its children, and complete this
// lookup unless a FULL Gloas child still awaits its payload (post-Gloas the payload
// envelope arrives separately from the block).
let bid_block_hash = self
.single_block_lookups
.get(&id)
.and_then(|lookup| lookup.peek_downloaded_bid_block_hash());
let import_action = match bid_block_hash {
Some(bid_block_hash) => ImportedAction::GloasBlockComplete {
block_root,
bid_block_hash,
},
None => ImportedAction::LookupComplete { block_root },
};
self.continue_child_lookups(import_action, cx);
if !self.has_any_awaiting_children(block_root) {
self.single_block_lookups.remove(&id);
metrics::inc_counter(&metrics::SYNC_LOOKUP_COMPLETED);
self.metrics.completed_lookups += 1;
debug!(
?block_root,
id, "Dropping completed lookup (external import)"
);
}
self.update_metrics();
} 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.
let result = {
let Some(lookup) = self.single_block_lookups.get_mut(&id) else {
return;
};
lookup.reset_requests();
lookup.continue_requests(cx)
};
self.on_lookup_result(id, result, "external_processing_result", cx);
}
}
pub fn has_any_awaiting_children(&self, block_root: Hash256) -> bool {
self.single_block_lookups
.iter()

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

@@ -243,13 +243,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;
self.payload_request = PayloadRequest::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

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

@@ -182,11 +182,6 @@ pub enum SyncMessage<E: EthSpec> {
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.
@@ -914,14 +909,6 @@ impl<T: BeaconChainTypes> SyncManager<T> {
} => 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(

34
beacon_node/network/src/sync/network_context.rs
View File

@@ -53,9 +53,8 @@ use task_executor::TaskExecutor;
use tokio::sync::mpsc;
use tracing::{Span, debug, debug_span, error, warn};
use types::{
BlobSidecar, BlockImportSource, ChainSpec, ColumnIndex, DataColumnSidecar,
DataColumnSidecarList, EthSpec, ForkContext, Hash256, SignedBeaconBlock,
SignedExecutionPayloadEnvelope, Slot,
BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar, DataColumnSidecarList, EthSpec,
ForkContext, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot,
};
pub mod custody;
@@ -855,26 +854,15 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
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 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) => {

107
beacon_node/network/src/sync/tests/lookups.rs
View File

@@ -1470,12 +1470,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!(
@@ -1565,15 +1559,6 @@ impl TestRig {
genesis_fork().fulu_enabled().then(Self::default)
}
fn new_after_deneb_before_fulu() -> Option<Self> {
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<Self> {
genesis_fork().fulu_enabled().then(|| {
Self::new(TestRigConfig {
@@ -1912,56 +1897,6 @@ impl TestRig {
}
}
fn insert_block_to_da_checker_as_pre_execution(&mut self, block: Arc<SignedBeaconBlock<E>>) {
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<SignedBeaconBlock<E>>,
) {
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 {
@@ -2584,48 +2519,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),+ $(,)?]