diff --git a/beacon_node/beacon_chain/src/inclusion_list_verification/gossip_verified_inclusion_list.rs b/beacon_node/beacon_chain/src/inclusion_list_verification/gossip_verified_inclusion_list.rs index 8b47421af55..6b89d571b64 100644 --- a/beacon_node/beacon_chain/src/inclusion_list_verification/gossip_verified_inclusion_list.rs +++ b/beacon_node/beacon_chain/src/inclusion_list_verification/gossip_verified_inclusion_list.rs @@ -1,7 +1,7 @@ use crate::canonical_head::CanonicalHead; use crate::inclusion_list_store::{InclusionListStore, InsertOutcome}; use crate::inclusion_list_verification::{ - InclusionListVerificationError, verify_inclusion_list_transactions_bounds, + InclusionListSource, InclusionListVerificationError, verify_inclusion_list_transactions_bounds, }; use crate::shuffling_cache::{ShufflingCache, with_cached_shuffling}; use crate::validator_pubkey_cache::ValidatorPubkeyCache; @@ -17,6 +17,7 @@ use tracing::debug; use types::{ChainSpec, EthSpec, Hash256, SignedInclusionList, Slot}; pub struct GossipVerificationContext<'a, T: BeaconChainTypes> { + pub source: InclusionListSource, pub canonical_head: &'a CanonicalHead, pub inclusion_list_store: &'a RwLock>, pub shuffling_cache: &'a RwLock>, @@ -46,21 +47,28 @@ impl GossipVerifiedInclusionList { let validator_index = inclusion_list.validator_index; let dependent_root = inclusion_list.dependent_root; - // [IGNORE] This is the first or second valid message from this validator. - if ctx - .inclusion_list_store - .read() - .seen_twice(slot, dependent_root, validator_index) - { - return Err(InclusionListVerificationError::AlreadySeenTwice { - validator_index, - slot, - dependent_root, - }); - } + match ctx.source { + InclusionListSource::Gossip => { + // [IGNORE] This is the first or second valid message from this validator. + if ctx + .inclusion_list_store + .read() + .seen_twice(slot, dependent_root, validator_index) + { + return Err(InclusionListVerificationError::AlreadySeenTwice { + validator_index, + slot, + dependent_root, + }); + } - // [IGNORE] `slot` is within the `MAXIMUM_GOSSIP_CLOCK_DISPARITY` allowance. - verify_propagation_slot_range(ctx.slot_clock, slot, ctx.spec)?; + // [IGNORE] `slot` is within the `MAXIMUM_GOSSIP_CLOCK_DISPARITY` allowance. + verify_propagation_slot_range(ctx.slot_clock, slot, ctx.spec)?; + } + InclusionListSource::Rpc => { + verify_retention_slot_range(ctx.slot_clock, slot, ctx.spec)? + } + } // [IGNORE] The size of the inclusion list transactions is non-zero. let transactions_size: u64 = inclusion_list @@ -205,9 +213,44 @@ fn verify_propagation_slot_range( Ok(()) } +/// Verify that the `slot` is within the range of slots the inclusion list store retains, +/// with reference to the current slot of the clock. +/// +/// Accounts for `MIN_SLOTS_FOR_INCLUSION_LISTS_REQUESTS`. +fn verify_retention_slot_range( + slot_clock: &S, + message_slot: Slot, + spec: &ChainSpec, +) -> Result<(), InclusionListVerificationError> { + let current_slot = slot_clock + .now() + .ok_or(InclusionListVerificationError::UnableToReadSlot)?; + if message_slot > current_slot { + return Err(InclusionListVerificationError::FutureSlot { + message_slot, + latest_permissible_slot: current_slot, + }); + } + + let earliest_permissible_slot = + current_slot.saturating_sub(spec.min_slots_for_inclusion_lists_requests); + if message_slot < earliest_permissible_slot { + return Err(InclusionListVerificationError::PastSlot { + message_slot, + earliest_permissible_slot, + }); + } + + Ok(()) +} + impl BeaconChain { - pub fn inclusion_list_gossip_verification_context(&self) -> GossipVerificationContext<'_, T> { + pub fn inclusion_list_gossip_verification_context( + &self, + source: InclusionListSource, + ) -> GossipVerificationContext<'_, T> { GossipVerificationContext { + source, canonical_head: &self.canonical_head, inclusion_list_store: &self.inclusion_list_store, shuffling_cache: &self.shuffling_cache, @@ -224,18 +267,20 @@ impl BeaconChain { &self, signed_inclusion_list: SignedInclusionList, seen_timestamp: Duration, + source: InclusionListSource, ) -> Result { metrics::inc_counter(&metrics::INCLUSION_LIST_PROCESSING_REQUESTS); let slot = signed_inclusion_list.message.slot; let validator_index = signed_inclusion_list.message.validator_index; - let ctx = self.inclusion_list_gossip_verification_context(); + let ctx = self.inclusion_list_gossip_verification_context(source); match GossipVerifiedInclusionList::new(signed_inclusion_list, seen_timestamp, &ctx) { Ok(verified) => { metrics::inc_counter(&metrics::INCLUSION_LIST_PROCESSING_SUCCESSES); debug!( %slot, %validator_index, + ?source, is_timely = verified.is_timely, "Successfully verified gossip inclusion list" ); @@ -247,6 +292,7 @@ impl BeaconChain { error = e.to_string(), %slot, %validator_index, + ?source, "Rejected gossip inclusion list" ); Err(e) diff --git a/beacon_node/beacon_chain/src/inclusion_list_verification/mod.rs b/beacon_node/beacon_chain/src/inclusion_list_verification/mod.rs index 3e2484e46b2..51377817549 100644 --- a/beacon_node/beacon_chain/src/inclusion_list_verification/mod.rs +++ b/beacon_node/beacon_chain/src/inclusion_list_verification/mod.rs @@ -16,6 +16,13 @@ use types::{BeaconStateError, ChainSpec, Hash256, ProgressiveTransactions, Slot} pub mod gossip_verified_inclusion_list; +/// The path through which an inclusion list reached this node. +#[derive(Debug, Clone, Copy, PartialEq, Eq, AsRefStr)] +pub enum InclusionListSource { + Gossip, + Rpc, +} + #[cfg(test)] mod tests; @@ -69,6 +76,32 @@ pub enum InclusionListVerificationError { BeaconStateError(BeaconStateError), } +impl InclusionListVerificationError { + /// Whether the peer that sent the inclusion list should be penalized. + pub fn penalize_peer(&self) -> bool { + // This match statement should never have a default case so that we are always forced to + // consider whether or not to penalize a peer when we add a new error condition. + match self { + // The list may be valid but is not useful to this node, or the node cannot tell. + InclusionListVerificationError::AlreadySeenTwice { .. } + | InclusionListVerificationError::FutureSlot { .. } + | InclusionListVerificationError::PastSlot { .. } + | InclusionListVerificationError::EmptyTransactions + | InclusionListVerificationError::DependentRootUnknown { .. } + | InclusionListVerificationError::InvalidDependentRoot { .. } + | InclusionListVerificationError::UnableToReadSlot + | InclusionListVerificationError::BeaconChainError(_) + | InclusionListVerificationError::BeaconStateError(_) => false, + // The list is invalid and an honest peer would not have sent it. + InclusionListVerificationError::InvalidTransactions(_) + | InclusionListVerificationError::DependentRootTooRecent { .. } + | InclusionListVerificationError::NotInCommittee { .. } + | InclusionListVerificationError::UnknownValidatorIndex(_) + | InclusionListVerificationError::InvalidSignature => true, + } + } +} + impl std::fmt::Display for InclusionListVerificationError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!(f, "{:?}", self) diff --git a/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs b/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs index 7bf24e61c5e..e93e06ac823 100644 --- a/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs +++ b/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs @@ -12,7 +12,7 @@ use crate::{ WhenSlotSkipped, inclusion_list_store::InsertOutcome, inclusion_list_verification::{ - InclusionListVerificationError, + InclusionListSource, InclusionListVerificationError, gossip_verified_inclusion_list::{GossipVerificationContext, GossipVerifiedInclusionList}, }, test_utils::{BeaconChainHarness, EphemeralHarnessType, fork_name_from_env, test_spec}, @@ -66,7 +66,13 @@ impl TestContext { fn gossip_ctx(&self) -> GossipVerificationContext<'_, T> { self.harness .chain - .inclusion_list_gossip_verification_context() + .inclusion_list_gossip_verification_context(InclusionListSource::Gossip) + } + + fn rpc_ctx(&self) -> GossipVerificationContext<'_, T> { + self.harness + .chain + .inclusion_list_gossip_verification_context(InclusionListSource::Rpc) } fn seen_timestamp(&self) -> Duration { @@ -588,3 +594,68 @@ async fn inclusion_list_in_first_heze_epoch() { let result = GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.gossip_ctx()); assert!(result.is_ok(), "expected Ok, got: {:?}", result); } + +#[test] +fn rpc_list_already_seen_twice_is_verified() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } + let ctx = TestContext::new(); + let slot = ctx.current_slot(); + + for tx in [0xaa, 0xbb] { + let signed = ctx.valid_inclusion_list(slot, ctx.genesis_block_root, vec![vec![tx]]); + let verified = + GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.gossip_ctx()) + .expect("should verify inclusion list"); + ctx.harness.chain.import_inclusion_list(verified); + } + + let signed = ctx.valid_inclusion_list(slot, ctx.genesis_block_root, vec![vec![0xcc]]); + assert!(GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.rpc_ctx()).is_ok()); +} + +#[test] +fn rpc_list_within_retention_window() { + if !fork_name_from_env().is_some_and(|f| f.gloas_enabled()) { + return; + } + let ctx = TestContext::new(); + let slot = ctx.current_slot(); + let signed = ctx.valid_inclusion_list(slot, ctx.genesis_block_root, vec![vec![0xaa]]); + + // A list for a slot the clock hasn't reached yet is rejected, the same as on gossip. + let future = ctx.valid_inclusion_list(slot + 1, ctx.genesis_block_root, vec![vec![0xaa]]); + assert!(matches!( + GossipVerifiedInclusionList::new(future, ctx.seen_timestamp(), &ctx.rpc_ctx()), + Err(InclusionListVerificationError::FutureSlot { .. }) + )); + + let spec = &ctx.harness.spec; + let last_retained_slot = slot + spec.min_slots_for_inclusion_lists_requests; + let first_unretained_slot = last_retained_slot + 1; + + // Move past the gossip clock disparity into the last retained slot. + let last_retained_slot_start = spec.get_slot_duration() * last_retained_slot.as_u64() as u32; + ctx.harness + .chain + .slot_clock + .set_current_time(last_retained_slot_start + spec.maximum_gossip_clock_disparity()); + assert!(matches!( + GossipVerifiedInclusionList::new(signed.clone(), ctx.seen_timestamp(), &ctx.gossip_ctx()), + Err(InclusionListVerificationError::PastSlot { .. }) + )); + assert!( + GossipVerifiedInclusionList::new(signed.clone(), ctx.seen_timestamp(), &ctx.rpc_ctx()) + .is_ok() + ); + + ctx.harness + .chain + .slot_clock + .set_slot(first_unretained_slot.as_u64()); + assert!(matches!( + GossipVerifiedInclusionList::new(signed, ctx.seen_timestamp(), &ctx.rpc_ctx()), + Err(InclusionListVerificationError::PastSlot { .. }) + )); +} diff --git a/beacon_node/beacon_chain/tests/events.rs b/beacon_node/beacon_chain/tests/events.rs index fb7dc94c18c..729d807e4bd 100644 --- a/beacon_node/beacon_chain/tests/events.rs +++ b/beacon_node/beacon_chain/tests/events.rs @@ -1,5 +1,6 @@ use arbitrary::Arbitrary; use beacon_chain::data_column_verification::GossipVerifiedDataColumn; +use beacon_chain::inclusion_list_verification::InclusionListSource; use beacon_chain::payload_envelope_verification::EnvelopeSource; use beacon_chain::test_utils::{ BeaconChainHarness, fork_name_from_env, generate_data_column_sidecars_from_block, test_spec, @@ -554,6 +555,7 @@ async fn inclusion_list_event_on_import() { .verify_inclusion_list_for_gossip( signed.clone(), harness.chain.slot_clock.now_duration().unwrap(), + InclusionListSource::Gossip, ) .expect("verification should succeed"); harness.chain.import_inclusion_list(verified); diff --git a/beacon_node/beacon_processor/src/lib.rs b/beacon_node/beacon_processor/src/lib.rs index c783c250642..44d6f056357 100644 --- a/beacon_node/beacon_processor/src/lib.rs +++ b/beacon_node/beacon_processor/src/lib.rs @@ -435,6 +435,7 @@ pub enum Work { }, RpcCustodyColumn(AsyncFn), RpcEnvelope(AsyncFn), + RpcInclusionLists(AsyncFn), ColumnReconstruction(AsyncFn), IgnoredRpcBlock { process_fn: BlockingFn, @@ -506,6 +507,7 @@ pub enum WorkType { RpcBlobs, RpcCustodyColumn, RpcEnvelope, + RpcInclusionLists, ColumnReconstruction, IgnoredRpcBlock, ChainSegment, @@ -574,6 +576,7 @@ impl Work { Work::RpcBlobs { .. } => WorkType::RpcBlobs, Work::RpcCustodyColumn { .. } => WorkType::RpcCustodyColumn, Work::RpcEnvelope(_) => WorkType::RpcEnvelope, + Work::RpcInclusionLists(_) => WorkType::RpcInclusionLists, Work::ColumnReconstruction(_) => WorkType::ColumnReconstruction, Work::IgnoredRpcBlock { .. } => WorkType::IgnoredRpcBlock, Work::ChainSegment { .. } => WorkType::ChainSegment, @@ -1004,6 +1007,8 @@ impl BeaconProcessor { // Check inclusion lists next, they also influence fork choice. } else if let Some(item) = work_queues.gossip_inclusion_list_queue.pop() { Some(item) + } else if let Some(item) = work_queues.rpc_inclusion_lists_queue.pop() { + Some(item) // Check sync committee messages after attestations as their rewards are lesser // and they don't influence fork choice. } else if let Some(item) = work_queues.sync_contribution_queue.pop() { @@ -1234,6 +1239,9 @@ impl BeaconProcessor { Work::RpcEnvelope(_) => { work_queues.rpc_envelope_queue.push(work, work_id) } + Work::RpcInclusionLists(_) => { + work_queues.rpc_inclusion_lists_queue.push(work, work_id) + } Work::RpcCustodyColumn { .. } => { work_queues.rpc_custody_column_queue.push(work, work_id) } @@ -1390,6 +1398,7 @@ impl BeaconProcessor { work_queues.rpc_blob_queue.len() } WorkType::RpcEnvelope => work_queues.rpc_envelope_queue.len(), + WorkType::RpcInclusionLists => work_queues.rpc_inclusion_lists_queue.len(), WorkType::RpcCustodyColumn => work_queues.rpc_custody_column_queue.len(), WorkType::ColumnReconstruction => { work_queues.column_reconstruction_queue.len() @@ -1595,6 +1604,7 @@ impl BeaconProcessor { | Work::RpcBlobs { process_fn } | Work::RpcCustodyColumn(process_fn) | Work::RpcEnvelope(process_fn) + | Work::RpcInclusionLists(process_fn) | Work::ColumnReconstruction(process_fn) => task_spawner.spawn_async(process_fn), Work::IgnoredRpcBlock { process_fn } => task_spawner.spawn_blocking(process_fn), Work::GossipBlock(work) diff --git a/beacon_node/beacon_processor/src/scheduler/work_queue.rs b/beacon_node/beacon_processor/src/scheduler/work_queue.rs index 10af51b3463..e720bfb9743 100644 --- a/beacon_node/beacon_processor/src/scheduler/work_queue.rs +++ b/beacon_node/beacon_processor/src/scheduler/work_queue.rs @@ -123,6 +123,7 @@ pub struct BeaconProcessorQueueLengths { rpc_blob_queue: usize, rpc_custody_column_queue: usize, rpc_envelope_queue: usize, + rpc_inclusion_lists_queue: usize, column_reconstruction_queue: usize, chain_segment_queue: usize, backfill_chain_segment: usize, @@ -206,6 +207,7 @@ impl BeaconProcessorQueueLengths { rpc_custody_column_queue: 64, // Bounded by `PARENT_DEPTH_TOLERANCE`; one envelope per Gloas block. rpc_envelope_queue: 1024, + rpc_inclusion_lists_queue: 64, column_reconstruction_queue: 1, chain_segment_queue: 64, backfill_chain_segment: 64, @@ -269,6 +271,7 @@ pub struct WorkQueues { pub rpc_blob_queue: FifoQueue>, pub rpc_custody_column_queue: FifoQueue>, pub rpc_envelope_queue: FifoQueue>, + pub rpc_inclusion_lists_queue: FifoQueue>, pub column_reconstruction_queue: LifoQueue>, pub chain_segment_queue: FifoQueue>, pub backfill_chain_segment: FifoQueue>, @@ -345,6 +348,7 @@ impl WorkQueues { let rpc_blob_queue = FifoQueue::new(queue_lengths.rpc_blob_queue); let rpc_custody_column_queue = FifoQueue::new(queue_lengths.rpc_custody_column_queue); let rpc_envelope_queue = FifoQueue::new(queue_lengths.rpc_envelope_queue); + let rpc_inclusion_lists_queue = FifoQueue::new(queue_lengths.rpc_inclusion_lists_queue); let column_reconstruction_queue = LifoQueue::new(queue_lengths.column_reconstruction_queue); let chain_segment_queue = FifoQueue::new(queue_lengths.chain_segment_queue); let backfill_chain_segment = FifoQueue::new(queue_lengths.backfill_chain_segment); @@ -418,6 +422,7 @@ impl WorkQueues { rpc_blob_queue, rpc_custody_column_queue, rpc_envelope_queue, + rpc_inclusion_lists_queue, chain_segment_queue, column_reconstruction_queue, backfill_chain_segment, diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs index 08e8e64bdf2..74869acdf35 100644 --- a/beacon_node/lighthouse_network/src/service/api_types.rs +++ b/beacon_node/lighthouse_network/src/service/api_types.rs @@ -10,6 +10,9 @@ use types::{ pub type Id = u32; +/// Id of an inclusion_lists_by_indices request sent by sync. +pub type InclusionListsByIndicesRequestId = Id; + #[derive(Debug, Hash, PartialEq, Eq, Clone, Copy)] pub struct SingleLookupReqId { pub lookup_id: Id, @@ -33,6 +36,10 @@ pub enum SyncRequestId { DataColumnsByRange(DataColumnsByRangeRequestId), /// Payload envelopes by range request PayloadEnvelopesByRange(PayloadEnvelopesByRangeRequestId), + /// Inclusion lists by indices request + InclusionListsByIndices { + id: InclusionListsByIndicesRequestId, + }, } /// Request ID for data_columns_by_root requests. Block lookups do not issue this request directly. diff --git a/beacon_node/network/src/metrics.rs b/beacon_node/network/src/metrics.rs index a0b84a2c706..358c0880cc1 100644 --- a/beacon_node/network/src/metrics.rs +++ b/beacon_node/network/src/metrics.rs @@ -516,6 +516,14 @@ pub static SYNC_DATA_COLUMNS_BY_RANGE_REQUEST_COLUMNS: LazyLock> = + LazyLock::new(|| { + try_create_histogram_with_buckets( + "sync_inclusion_lists_by_indices_request_indices", + "Number of inclusion list committee positions requested from one peer in a single InclusionListsByIndices request", + Ok(vec![1.0, 2.0, 4.0, 8.0, 16.0]), + ) + }); pub static SYNCING_CHAIN_BATCH_DOWNLOADING: LazyLock> = LazyLock::new(|| { try_create_histogram_with_buckets( "sync_range_chain_batch_downloading_seconds", 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 ee537c0e8ba..d5ea71f4818 100644 --- a/beacon_node/network/src/network_beacon_processor/gossip_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/gossip_methods.rs @@ -12,7 +12,9 @@ use beacon_chain::data_column_verification::{ use beacon_chain::execution_proof_verification::{Error as ExecutionProofError, ProofSource}; use beacon_chain::fetch_blobs::PartialHeaderOrBid; use beacon_chain::inclusion_list_store::InsertOutcome; -use beacon_chain::inclusion_list_verification::InclusionListVerificationError; +use beacon_chain::inclusion_list_verification::{ + InclusionListSource, InclusionListVerificationError, +}; use beacon_chain::partial_data_column_assembler::UpdatedPartials; use beacon_chain::payload_bid_verification::PayloadBidError; use beacon_chain::payload_envelope_verification::{ @@ -4372,9 +4374,11 @@ impl NetworkBeaconProcessor { inclusion_list: Box, seen_timestamp: Duration, ) { - let verification_result = self - .chain - .verify_inclusion_list_for_gossip(*inclusion_list, seen_timestamp); + let verification_result = self.chain.verify_inclusion_list_for_gossip( + *inclusion_list, + seen_timestamp, + InclusionListSource::Gossip, + ); if let Err(error) = &verification_result { metrics::register_inclusion_list_error(error); } diff --git a/beacon_node/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index 8fefd32d0fd..0b6ea6b15f3 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -608,6 +608,22 @@ impl NetworkBeaconProcessor { }) } + /// Create a new `Work` event for inclusion lists fetched via RPC. + pub fn send_rpc_inclusion_lists( + self: &Arc, + peer_id: PeerId, + inclusion_lists: Vec>, + ) -> Result<(), Error> { + let s = self.clone(); + self.try_send(BeaconWorkEvent { + drop_during_sync: false, + work: Work::RpcInclusionLists(Box::pin(async move { + s.process_rpc_inclusion_lists(peer_id, inclusion_lists) + .await; + })), + }) + } + /// Create a new `Work` event for some custody columns. `process_rpc_custody_columns` reports /// the result back to sync. pub fn send_rpc_custody_columns( 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 df5939d697d..07553b62f57 100644 --- a/beacon_node/network/src/network_beacon_processor/sync_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/sync_methods.rs @@ -1,5 +1,6 @@ use crate::metrics::{self, register_process_result_metrics}; use crate::network_beacon_processor::{FUTURE_SLOT_TOLERANCE, NetworkBeaconProcessor}; +use crate::service::NetworkMessage; use crate::sync::BatchProcessResult; use crate::sync::manager::CustodyBatchProcessResult; use crate::sync::{ @@ -13,6 +14,7 @@ use beacon_chain::data_availability_checker::{ }; use beacon_chain::fetch_blobs::PartialHeaderOrBid; use beacon_chain::historical_data_columns::HistoricalDataColumnError; +use beacon_chain::inclusion_list_verification::InclusionListSource; use beacon_chain::payload_envelope_verification::EnvelopeSource; use beacon_chain::{ AvailabilityProcessingStatus, BeaconChainTypes, BlockError, ChainSegmentResult, @@ -25,8 +27,10 @@ use beacon_processor::{ use beacon_processor::{Work, WorkEvent}; use lighthouse_network::PeerAction; use lighthouse_network::PeerId; +use lighthouse_network::ReportSource; use lighthouse_network::service::api_types::CustodyBackfillBatchId; use logging::crit; +use slot_clock::SlotClock; use std::sync::Arc; use tracing::{debug, debug_span, error, info, instrument, warn}; use types::{BlockImportSource, DataColumnSidecarList, Epoch, ExecutionBlockHash, Hash256}; @@ -303,6 +307,52 @@ impl NetworkBeaconProcessor { }); } + /// Attempt to verify and import inclusion lists received via RPC. + #[instrument( + name = "lh_process_rpc_inclusion_lists", + parent = None, + level = "debug", + skip_all, + fields(%peer_id), + )] + pub async fn process_rpc_inclusion_lists( + self: Arc>, + peer_id: PeerId, + inclusion_lists: Vec>, + ) { + debug!( + count = inclusion_lists.len(), + "Processing RPC inclusion lists" + ); + + let seen_timestamp = self.chain.slot_clock.now_duration().unwrap_or_default(); + for inclusion_list in inclusion_lists { + let slot = inclusion_list.message.slot; + let validator_index = inclusion_list.message.validator_index; + match self.chain.verify_inclusion_list_for_gossip( + Arc::unwrap_or_clone(inclusion_list), + seen_timestamp, + InclusionListSource::Rpc, + ) { + Ok(verified_inclusion_list) => { + let outcome = self.chain.import_inclusion_list(verified_inclusion_list); + debug!(%slot, %validator_index, ?outcome, "Imported RPC inclusion list"); + } + Err(e) if e.penalize_peer() => { + self.send_network_message(NetworkMessage::ReportPeer { + peer_id, + action: PeerAction::LowToleranceError, + source: ReportSource::SyncService, + msg: "invalid_rpc_inclusion_list", + }); + } + // The list is not useful to this node or could not be verified. + // The rejection is logged by the verifier, and the peer is not at fault. + Err(_) => {} + } + } + } + /// Attempt to verify and import an execution payload envelope received via RPC. #[instrument( name = "lh_process_lookup_envelope", diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index 2212855ad0d..174e7d5379b 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -3377,3 +3377,96 @@ async fn test_gossip_inclusion_list_rejection_penalises_the_peer() { } ); } + +// A valid list fetched over RPC is imported into the store without penalizing the peer. +#[tokio::test] +async fn test_rpc_inclusion_list_is_imported() { + let mut rig = TestRig::new(SMALL_CHAIN).await; + let slot = rig.chain.slot().unwrap(); + let (committee, dependent_root) = rig + .chain + .inclusion_list_committee(rig.chain.head_beacon_block_root(), slot) + .unwrap(); + let validator_index = committee[0]; + + let message = InclusionList { + slot, + validator_index, + dependent_root, + transactions: vec![vec![0xaa].try_into().unwrap()].try_into().unwrap(), + }; + let epoch = slot.epoch(E::slots_per_epoch()); + let domain = rig.chain.spec.get_domain( + epoch, + Domain::InclusionListCommittee, + &rig.chain.spec.fork_at_epoch(epoch), + rig.chain.genesis_validators_root, + ); + let signature = rig._harness.validator_keypairs[validator_index as usize] + .sk + .sign(message.signing_root(domain)); + let inclusion_list = SignedInclusionList { message, signature }; + + rig.network_beacon_processor + .send_rpc_inclusion_lists(junk_peer_id(), vec![Arc::new(inclusion_list.clone())]) + .unwrap(); + + let stored_inclusion_lists = || { + rig.chain + .inclusion_list_store + .read() + .get_signed_inclusion_lists(slot, dependent_root, &[validator_index]) + }; + let result = tokio::time::timeout(STANDARD_TIMEOUT, async { + while stored_inclusion_lists().is_empty() { + tokio::time::sleep(Duration::from_millis(20)).await; + } + }) + .await; + assert!( + result.is_ok(), + "timed out waiting for the list to be imported" + ); + assert_eq!(stored_inclusion_lists(), vec![inclusion_list]); + assert!( + rig.receive_network_messages_with_timeout(Duration::from_millis(100), None) + .await + .is_none(), + "a valid list should not penalize the peer" + ); +} + +// A list fetched over RPC from a validator outside the committee +// penalizes the peer and is not imported. +#[tokio::test] +async fn test_rpc_inclusion_list_outside_the_committee_penalizes_the_peer() { + let mut rig = TestRig::new(SMALL_CHAIN).await; + let slot = rig.chain.slot().unwrap(); + let (_, dependent_root) = rig + .chain + .inclusion_list_committee(rig.chain.head_beacon_block_root(), slot) + .unwrap(); + let validator_index = u64::MAX; + let inclusion_list = signed_inclusion_list(slot, validator_index, dependent_root, 0xaa); + + rig.network_beacon_processor + .send_rpc_inclusion_lists(junk_peer_id(), vec![Arc::new(inclusion_list)]) + .unwrap(); + + let network_message = rig + .receive_network_messages_with_timeout(Duration::from_secs(1), Some(1)) + .await + .and_then(|mut messages| messages.pop()) + .expect("should penalize the peer"); + match network_message { + NetworkMessage::ReportPeer { msg, .. } => assert_eq!(msg, "invalid_rpc_inclusion_list"), + other => panic!("expected ReportPeer, got {:?}", other), + } + assert!( + rig.chain + .inclusion_list_store + .read() + .get_signed_inclusion_lists(slot, dependent_root, &[validator_index]) + .is_empty() + ); +} diff --git a/beacon_node/network/src/service.rs b/beacon_node/network/src/service.rs index 6ace7bdbf68..26ef7f05321 100644 --- a/beacon_node/network/src/service.rs +++ b/beacon_node/network/src/service.rs @@ -46,8 +46,8 @@ use tracing::{debug, error, info, trace, warn}; use typenum::Unsigned; use types::{ BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, PartialDataColumn, SignedBeaconBlock, - SignedExecutionPayloadEnvelope, SubnetId, SyncCommitteeSubscription, SyncSubnetId, - ValidatorSubscription, + SignedExecutionPayloadEnvelope, SignedInclusionList, SubnetId, SyncCommitteeSubscription, + SyncSubnetId, ValidatorSubscription, }; mod tests; @@ -1167,14 +1167,18 @@ impl NetworkService { Response::PayloadEnvelopesByRange(envelope) => { self.on_payload_envelopes_by_range_response(peer_id, app_request_id, envelope); } + Response::InclusionListsByIndices(inclusion_list) => { + self.on_inclusion_lists_by_indices_response( + peer_id, + app_request_id, + inclusion_list, + ); + } // Lighthouse currently only serves BlocksByHead and does not issue it as a client, // so receiving a response is unexpected. Drop it without crashing. Response::BlocksByHead(_) => { debug!("BlocksByHead response received but not requested by lighthouse"); } - Response::InclusionListsByIndices(_) => { - debug!("InclusionListsByIndices response received but not requested by lighthouse"); - } // Light client responses should not be received Response::LightClientBootstrap(_) | Response::LightClientOptimisticUpdate(_) @@ -1375,6 +1379,29 @@ impl NetworkService { }); } + /// Handle an `InclusionListsByIndices` response from the peer. + fn on_inclusion_lists_by_indices_response( + &mut self, + peer_id: PeerId, + app_request_id: AppRequestId, + inclusion_list: Option>, + ) { + trace!( + %peer_id, + "Received InclusionListsByIndices Response" + ); + + if let AppRequestId::Sync(sync_request_id) = app_request_id { + self.send_to_sync(SyncMessage::RpcInclusionList { + peer_id, + sync_request_id, + inclusion_list, + }); + } else { + crit!("All inclusion lists by indices responses should belong to sync"); + } + } + /// Sends a message to the sync manager. fn send_to_sync(&mut self, message: SyncMessage) { self.sync_send.send(message).unwrap_or_else(|e| { diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index dae965a6d16..650d2569d3e 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -47,8 +47,12 @@ use crate::service::NetworkMessage; use crate::status::ToStatusMessage; use crate::sync::block_lookups::{BlockComponent, DownloadResult}; use crate::sync::custody_backfill_sync::CustodyBackFillSync; -use crate::sync::network_context::{PeerGroup, RpcResponseResult}; +use crate::sync::network_context::{ + InclusionListCommitteePositions, InclusionListsByIndicesRequestParams, PeerGroup, + RpcResponseResult, +}; use beacon_chain::block_verification_types::AsBlock; +use beacon_chain::inclusion_list_store::DependentRoot; use beacon_chain::{BeaconChain, BeaconChainTypes, EngineState}; use futures::StreamExt; use lighthouse_network::SyncInfo; @@ -57,8 +61,8 @@ use lighthouse_network::service::api_types::{ BlobsByRangeRequestId, BlocksByRangeRequestId, ComponentsByRangeRequestId, CustodyBackFillBatchRequestId, CustodyBackfillBatchId, CustodyRequester, DataColumnsByRangeRequestId, DataColumnsByRangeRequester, DataColumnsByRootRequestId, - DataColumnsByRootRequester, Id, PayloadEnvelopesByRangeRequestId, SingleLookupReqId, - SyncRequestId, + DataColumnsByRootRequester, Id, InclusionListsByIndicesRequestId, + PayloadEnvelopesByRangeRequestId, SingleLookupReqId, SyncRequestId, }; use lighthouse_network::types::{NetworkGlobals, SyncState}; use lighthouse_network::{PeerAction, PeerId}; @@ -73,7 +77,7 @@ use tokio::sync::mpsc; use tracing::{debug, error, info, trace}; use types::{ BlobSidecar, DataColumnSidecar, EthSpec, ExecutionBlockHash, ForkContext, Hash256, - SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, + SignedBeaconBlock, SignedExecutionPayloadEnvelope, SignedInclusionList, Slot, }; /// The number of slots ahead of us that is allowed before requesting a long-range (batch) Sync @@ -136,6 +140,13 @@ pub enum SyncMessage { envelope: Option>>, }, + /// An inclusion list has been received from the RPC. + RpcInclusionList { + sync_request_id: SyncRequestId, + peer_id: PeerId, + inclusion_list: Option>, + }, + /// A block with an unknown parent has been received. UnknownParentBlock(PeerId, Arc>, Hash256), @@ -520,6 +531,9 @@ impl SyncManager { } SyncRequestId::PayloadEnvelopesByRange(req_id) => self .on_payload_envelopes_by_range_response(req_id, peer_id, RpcEvent::RPCError(error)), + SyncRequestId::InclusionListsByIndices { id } => { + self.on_inclusion_lists_by_indices_response(id, peer_id, RpcEvent::RPCError(error)) + } } } @@ -785,6 +799,11 @@ impl SyncManager { self.chain.slot_clock.slot_duration().as_secs() * T::EthSpec::slots_per_epoch(); let mut epoch_interval = tokio::time::interval(Duration::from_secs(epoch_duration)); + // Fires at the inclusion list deadline of every slot to request missing inclusion lists. + let inclusion_list_deadline = + tokio::time::sleep(self.duration_to_next_inclusion_list_deadline()); + tokio::pin!(inclusion_list_deadline); + // process any inbound messages loop { tokio::select! { @@ -809,10 +828,124 @@ impl SyncManager { _ = epoch_interval.tick() => { self.update_sync_state(); } + _ = &mut inclusion_list_deadline => { + self.on_inclusion_list_deadline().await; + inclusion_list_deadline.as_mut().reset( + tokio::time::Instant::now() + self.duration_to_next_inclusion_list_deadline(), + ); + } } } } + /// Time until the inclusion list deadline of the next slot. + fn duration_to_next_inclusion_list_deadline(&self) -> Duration { + let slot_clock = &self.chain.slot_clock; + slot_clock + .duration_to_next_slot() + .unwrap_or_else(|| slot_clock.slot_duration()) + + self.chain.spec.get_inclusion_list_due() + } + + /// Requests the inclusion lists this node is missing for the current slot when one of its + /// validators proposes the next slot, so the proposer can include their transactions. + pub(crate) async fn on_inclusion_list_deadline(&mut self) { + if !self.network_globals().sync_state.read().is_synced() { + return; + } + + let Ok(slot) = self.chain.slot() else { + return; + }; + + if !self + .chain + .spec + .fork_name_at_slot::(slot) + .heze_enabled() + { + return; + } + + if !self.has_proposer_at(slot + 1).await { + return; + } + + let Some((dependent_root, requested)) = self.get_missing_inclusion_lists(slot) else { + return; + }; + + if requested.is_empty() { + return; + } + + let request = InclusionListsByIndicesRequestParams { + slot, + dependent_root, + requested, + }; + if let Err(e) = self.network.inclusion_lists_by_indices_request(request) { + debug!(%slot, error = ?e, "Failed to request inclusion lists"); + } + } + + /// Returns whether this node has one of its validators proposing at `slot`. + async fn has_proposer_at(&self, slot: Slot) -> bool { + let cached_head = self.chain.canonical_head.cached_head(); + let head_state = &cached_head.snapshot.beacon_state; + + let Some(proposer_index) = head_state.proposer_lookahead().ok().and_then(|lookahead| { + let current_epoch_start_slot = head_state + .current_epoch() + .start_slot(T::EthSpec::slots_per_epoch()); + let offset = slot + .as_usize() + .checked_sub(current_epoch_start_slot.as_usize())?; + lookahead.get(offset).copied() + }) else { + return false; + }; + + let Some(execution_layer) = self.chain.execution_layer.as_ref() else { + return false; + }; + execution_layer + .has_proposer_preparation_data(proposer_index) + .await + } + + /// The inclusion list committee positions at `slot` whose inclusion list this node has not + /// observed yet, each with the validator at that position, and the dependent root of the + /// inclusion list committee. + fn get_missing_inclusion_lists( + &self, + slot: Slot, + ) -> Option<(DependentRoot, InclusionListCommitteePositions)> { + let head_block_root = self.chain.canonical_head.cached_head().head_block_root(); + + let (committee, dependent_root) = self + .chain + .inclusion_list_committee(head_block_root, slot) + .inspect_err( + |e| debug!(%slot, error = ?e, "Unable to read the inclusion list committee"), + ) + .ok()?; + let observed = self + .chain + .get_inclusion_list_bits(head_block_root, slot, false) + .inspect_err(|e| debug!(%slot, error = ?e, "Unable to read the inclusion list bits")) + .ok()?; + + let missing = committee + .iter() + .enumerate() + .filter(|(position, _)| !observed.get(*position).unwrap_or(false)) + .map(|(position, validator_index)| (position, *validator_index)) + .collect(); + + Some((dependent_root, missing)) + } + pub(crate) fn handle_message(&mut self, sync_message: SyncMessage) { match sync_message { SyncMessage::AddPeer(peer_id, info) => { @@ -854,6 +987,11 @@ impl SyncManager { peer_id, envelope, } => self.rpc_payload_envelope_received(sync_request_id, peer_id, envelope), + SyncMessage::RpcInclusionList { + sync_request_id, + peer_id, + inclusion_list, + } => self.rpc_inclusion_list_received(sync_request_id, peer_id, inclusion_list), SyncMessage::UnknownParentBlock(peer_id, block, block_root) => { let block_slot = block.slot(); let parent_root = block.parent_root(); @@ -1253,6 +1391,25 @@ impl SyncManager { } } + fn rpc_inclusion_list_received( + &mut self, + sync_request_id: SyncRequestId, + peer_id: PeerId, + inclusion_list: Option>, + ) { + match sync_request_id { + SyncRequestId::InclusionListsByIndices { id } => self + .on_inclusion_lists_by_indices_response( + id, + peer_id, + RpcEvent::from_chunk(inclusion_list), + ), + _ => { + crit!(%peer_id, "bad request id for inclusion list"); + } + } + } + fn on_single_payload_envelope_response( &mut self, id: SingleLookupReqId, @@ -1491,4 +1648,27 @@ impl SyncManager { } } } + + /// Handles receiving a response for an inclusion lists by indices request. + fn on_inclusion_lists_by_indices_response( + &mut self, + request_id: InclusionListsByIndicesRequestId, + peer_id: PeerId, + inclusion_list: RpcEvent>, + ) { + let Some(Ok(inclusion_lists)) = self.network.on_inclusion_lists_by_indices_response( + request_id, + peer_id, + inclusion_list, + ) else { + return; + }; + + if let Err(e) = self + .network + .send_inclusion_lists_for_processing(peer_id, inclusion_lists) + { + debug!(%peer_id, error = ?e, "Failed to send inclusion lists for processing"); + } + } } diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index c1f6b314918..e0064d23447 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -3,7 +3,8 @@ use self::custody::{ActiveCustodyRequest, Error as CustodyRequestError}; pub use self::requests::{ - BlocksByRootSingleRequest, DataColumnsByRootRequestParams, PayloadEnvelopesByRootSingleRequest, + BlocksByRootSingleRequest, DataColumnsByRootRequestParams, InclusionListCommitteePositions, + InclusionListsByIndicesRequestParams, PayloadEnvelopesByRootSingleRequest, }; use super::SyncMessage; use super::block_sidecar_coupling::RangeBlockComponentsRequest; @@ -34,8 +35,8 @@ use lighthouse_network::service::api_types::{ AppRequestId, BlobsByRangeRequestId, BlocksByRangeRequestId, ComponentsByRangeRequestId, CustodyBackFillBatchRequestId, CustodyBackfillBatchId, CustodyId, CustodyRequester, DataColumnsByRangeRequestId, DataColumnsByRangeRequester, DataColumnsByRootRequestId, - DataColumnsByRootRequester, Id, PayloadEnvelopesByRangeRequestId, SingleLookupReqId, - SyncRequestId, + DataColumnsByRootRequester, Id, InclusionListsByIndicesRequestId, + PayloadEnvelopesByRangeRequestId, SingleLookupReqId, SyncRequestId, }; use lighthouse_network::{Client, NetworkGlobals, PeerAction, PeerId, ReportSource}; use parking_lot::RwLock; @@ -43,7 +44,8 @@ pub use requests::LookupVerifyError; use requests::{ ActiveRequestItems, ActiveRequests, BlobsByRangeRequestItems, BlocksByRangeRequestItems, BlocksByRootRequestItems, DataColumnsByRangeRequestItems, DataColumnsByRootRequestItems, - PayloadEnvelopesByRangeRequestItems, PayloadEnvelopesByRootRequestItems, + InclusionListsByIndicesRequestItems, PayloadEnvelopesByRangeRequestItems, + PayloadEnvelopesByRootRequestItems, }; #[cfg(test)] use slot_clock::SlotClock; @@ -56,7 +58,8 @@ use tokio::sync::mpsc; use tracing::{Span, debug, debug_span, error, warn}; use types::{ BlobSidecar, ChainSpec, ColumnIndex, DataColumnSidecar, DataColumnSidecarList, Epoch, EthSpec, - ForkContext, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, + ForkContext, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, SignedInclusionList, + Slot, }; pub mod custody; @@ -142,6 +145,7 @@ pub enum RpcRequestSendError { pub enum NoPeerError { BlockPeer, CustodyPeer(ColumnIndex), + InclusionListPeer, } #[derive(Debug, PartialEq, Eq)] @@ -255,6 +259,10 @@ pub struct SyncNetworkContext { custody_backfill_data_column_batch_requests: FnvHashMap>, + /// A mapping of active InclusionListsByIndices requests + inclusion_lists_by_indices_requests: + ActiveRequests, + /// Whether the ee is online. If it's not, we don't allow access to the /// `beacon_processor_send`. execution_engine_state: EngineState, @@ -337,6 +345,7 @@ impl SyncNetworkContext { custody_by_root_requests: <_>::default(), components_by_range_requests: FnvHashMap::default(), custody_backfill_data_column_batch_requests: FnvHashMap::default(), + inclusion_lists_by_indices_requests: ActiveRequests::new("inclusion_lists_by_indices"), network_beacon_processor, chain, fork_context, @@ -367,6 +376,7 @@ impl SyncNetworkContext { blobs_by_range_requests, data_columns_by_range_requests, payload_envelopes_by_range_requests, + inclusion_lists_by_indices_requests, // custody_by_root_requests is a meta request of data_columns_by_root_requests custody_by_root_requests: _, // components_by_range_requests is a meta request of various _by_range requests @@ -406,6 +416,10 @@ impl SyncNetworkContext { .active_requests_of_peer(peer_id) .into_iter() .map(|req_id| SyncRequestId::PayloadEnvelopesByRange(*req_id)); + let inclusion_lists_by_indices_ids = inclusion_lists_by_indices_requests + .active_requests_of_peer(peer_id) + .into_iter() + .map(|id| SyncRequestId::InclusionListsByIndices { id: *id }); blocks_by_root_ids .chain(payload_envelopes_by_root_ids) .chain(data_column_by_root_ids) @@ -413,6 +427,7 @@ impl SyncNetworkContext { .chain(blobs_by_range_ids) .chain(data_column_by_range_ids) .chain(payload_envelope_by_range_ids) + .chain(inclusion_lists_by_indices_ids) .collect() } @@ -1224,6 +1239,79 @@ impl SyncNetworkContext { Ok(id) } + /// Requests inclusion lists from a synced peer. + pub fn inclusion_lists_by_indices_request( + &mut self, + request: InclusionListsByIndicesRequestParams, + ) -> Result { + let inclusion_lists_by_indices_per_peer = + ActiveRequestsPerPeer::new(&self.inclusion_lists_by_indices_requests); + let Some(peer_id) = self + .network_globals() + .peers + .read() + .synced_peers() + .map(|peer| { + ( + // Strictly de-prioritize peers already at the per-protocol concurrency limit + inclusion_lists_by_indices_per_peer.at_concurrency_limit(peer), + // Random factor to break ties, otherwise the PeerID breaks ties + rand::random::(), + *peer, + ) + }) + .min() + .map(|(_, _, peer)| peer) + else { + return Err(RpcRequestSendError::NoPeer(NoPeerError::InclusionListPeer)); + }; + + let id = self.next_id(); + let request_span = debug_span!( + parent: None, + "lh_outgoing_inclusion_lists_by_indices_request", + slot = %request.slot, + ); + + self.send_network_msg(NetworkMessage::SendRequest { + peer_id, + request: RequestType::InclusionListsByIndices( + request + .clone() + .try_into_request::() + .map_err(|e| RpcRequestSendError::InternalError(e.to_owned()))?, + ), + app_request_id: AppRequestId::Sync(SyncRequestId::InclusionListsByIndices { id }), + }) + .map_err(|_| RpcRequestSendError::InternalError("network send error".to_owned()))?; + + metrics::observe( + &metrics::SYNC_INCLUSION_LISTS_BY_INDICES_REQUEST_INDICES, + request.requested.len() as f64, + ); + + debug!( + method = "InclusionListsByIndices", + slot = %request.slot, + dependent_root = ?request.dependent_root, + requested = ?request.requested, + peer = %peer_id, + %id, + "Sync RPC request sent" + ); + + self.inclusion_lists_by_indices_requests.insert( + id, + peer_id, + // do not enforce that every requested list is returned, since a peer may hold fewer + // inclusion lists than requested + false, + InclusionListsByIndicesRequestItems::new(request), + request_span, + ); + Ok(id) + } + pub fn is_execution_engine_online(&self) -> bool { self.execution_engine_state == EngineState::Online } @@ -1452,6 +1540,18 @@ impl SyncNetworkContext { self.on_rpc_response_result(resp, peer_id) } + pub(crate) fn on_inclusion_lists_by_indices_response( + &mut self, + id: InclusionListsByIndicesRequestId, + peer_id: PeerId, + rpc_event: RpcEvent>, + ) -> Option>>> { + let resp = self + .inclusion_lists_by_indices_requests + .on_response(id, rpc_event); + self.on_rpc_response_result(resp, peer_id) + } + /// Common handler for consistent scoring of RpcResponseError fn on_rpc_response_result( &mut self, @@ -1575,6 +1675,29 @@ impl SyncNetworkContext { }) } + pub fn send_inclusion_lists_for_processing( + &self, + peer_id: PeerId, + inclusion_lists: Vec>, + ) -> Result<(), SendErrorProcessor> { + let beacon_processor = self + .beacon_processor_if_enabled() + .ok_or(SendErrorProcessor::ProcessorNotAvailable)?; + + debug!( + %peer_id, + count = inclusion_lists.len(), + "Sending inclusion lists for processing" + ); + + beacon_processor + .send_rpc_inclusion_lists(peer_id, inclusion_lists) + .map_err(|e| { + error!(error = ?e, "Failed to send sync inclusion lists to processor"); + SendErrorProcessor::SendError + }) + } + pub fn send_custody_columns_for_processing( &self, _id: Id, @@ -1700,6 +1823,11 @@ impl SyncNetworkContext { self.custody_backfill_data_column_batch_requests.len() } + #[cfg(test)] + pub(crate) fn inclusion_lists_by_indices_request_count(&self) -> usize { + self.inclusion_lists_by_indices_requests.len() + } + #[cfg(test)] pub(crate) fn insert_test_custody_backfill_entry( &mut self, @@ -1733,6 +1861,10 @@ impl SyncNetworkContext { "components_by_range", self.components_by_range_requests.len(), ), + ( + "inclusion_lists_by_indices", + self.inclusion_lists_by_indices_requests.len(), + ), ] { metrics::set_gauge_vec(&metrics::SYNC_ACTIVE_NETWORK_REQUESTS, &[id], count as i64); } diff --git a/beacon_node/network/src/sync/network_context/requests.rs b/beacon_node/network/src/sync/network_context/requests.rs index 18e46be07e7..ade3d5cd48e 100644 --- a/beacon_node/network/src/sync/network_context/requests.rs +++ b/beacon_node/network/src/sync/network_context/requests.rs @@ -12,6 +12,10 @@ pub use blocks_by_range::BlocksByRangeRequestItems; pub use blocks_by_root::{BlocksByRootRequestItems, BlocksByRootSingleRequest}; pub use data_columns_by_range::DataColumnsByRangeRequestItems; pub use data_columns_by_root::{DataColumnsByRootRequestItems, DataColumnsByRootRequestParams}; +pub use inclusion_lists_by_indices::{ + InclusionListCommitteePositions, InclusionListsByIndicesRequestItems, + InclusionListsByIndicesRequestParams, +}; pub use payload_envelopes_by_range::PayloadEnvelopesByRangeRequestItems; pub use payload_envelopes_by_root::{ PayloadEnvelopesByRootRequestItems, PayloadEnvelopesByRootSingleRequest, @@ -26,6 +30,7 @@ mod blocks_by_range; mod blocks_by_root; mod data_columns_by_range; mod data_columns_by_root; +mod inclusion_lists_by_indices; mod payload_envelopes_by_range; mod payload_envelopes_by_root; @@ -34,6 +39,7 @@ pub enum LookupVerifyError { NotEnoughResponsesReturned { actual: usize }, TooManyResponses, UnrequestedBlockRoot(Hash256), + UnrequestedDependentRoot(Hash256), UnrequestedIndex(u64), UnrequestedSlot(Slot), InvalidInclusionProof, diff --git a/beacon_node/network/src/sync/network_context/requests/inclusion_lists_by_indices.rs b/beacon_node/network/src/sync/network_context/requests/inclusion_lists_by_indices.rs new file mode 100644 index 00000000000..7160d05837c --- /dev/null +++ b/beacon_node/network/src/sync/network_context/requests/inclusion_lists_by_indices.rs @@ -0,0 +1,234 @@ +use lighthouse_network::rpc::methods::InclusionListsByIndicesRequest; +use std::collections::HashSet; +use std::sync::Arc; +use types::{EthSpec, Hash256, InclusionListBits, SignedInclusionList, Slot}; + +use super::{ActiveRequestItems, LookupVerifyError}; + +/// Inclusion list committee positions, each with the validator at that position. +pub type InclusionListCommitteePositions = Vec<(usize, u64)>; + +#[derive(Debug, Clone)] +pub struct InclusionListsByIndicesRequestParams { + pub slot: Slot, + pub dependent_root: Hash256, + /// Requested inclusion list committee positions, each with the validator index + /// corresponding to that position. + pub requested: InclusionListCommitteePositions, +} + +impl InclusionListsByIndicesRequestParams { + pub fn try_into_request( + self, + ) -> Result, &'static str> { + let mut indices = InclusionListBits::::new(); + for (position, _) in &self.requested { + indices + .set(*position, true) + .map_err(|_| "Position exceeds the inclusion list committee size")?; + } + Ok(InclusionListsByIndicesRequest { + slot: self.slot, + dependent_root: self.dependent_root, + indices, + }) + } +} + +pub struct InclusionListsByIndicesRequestItems { + request: InclusionListsByIndicesRequestParams, + /// Distinct validators in `request`: a validator holding multiple inclusion list + /// committee positions is served a single list. + requested_validators: HashSet, + items: Vec>, +} + +impl InclusionListsByIndicesRequestItems { + pub fn new(request: InclusionListsByIndicesRequestParams) -> Self { + let requested_validators = request + .requested + .iter() + .map(|(_, validator_index)| *validator_index) + .collect(); + Self { + request, + requested_validators, + items: vec![], + } + } +} + +impl ActiveRequestItems for InclusionListsByIndicesRequestItems { + type Item = Arc; + + /// Appends a chunk to this multi-item request. Returns `true` once an inclusion list has been + /// received for every requested validator, resolving the request before the stream terminator. + fn add(&mut self, inclusion_list: Self::Item) -> Result { + let slot = inclusion_list.message.slot; + if slot != self.request.slot { + return Err(LookupVerifyError::UnrequestedSlot(slot)); + } + + let dependent_root = inclusion_list.message.dependent_root; + if dependent_root != self.request.dependent_root { + return Err(LookupVerifyError::UnrequestedDependentRoot(dependent_root)); + } + + let validator_index = inclusion_list.message.validator_index; + if !self.requested_validators.contains(&validator_index) { + return Err(LookupVerifyError::UnrequestedIndex(validator_index)); + } + + if self + .items + .iter() + .any(|item| item.message.validator_index == validator_index) + { + return Err(LookupVerifyError::DuplicatedData(slot, validator_index)); + } + + self.items.push(inclusion_list); + + Ok(self.items.len() >= self.requested_validators.len()) + } + + fn consume(&mut self) -> Vec { + std::mem::take(&mut self.items) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bls::Signature; + use types::{InclusionList, MinimalEthSpec as E, ProgressiveTransactions}; + + const SLOT: Slot = Slot::new(1); + const DEPENDENT_ROOT: Hash256 = Hash256::repeat_byte(1); + + fn inclusion_list( + slot: Slot, + dependent_root: Hash256, + validator_index: u64, + ) -> Arc { + Arc::new(SignedInclusionList { + message: InclusionList { + slot, + validator_index, + dependent_root, + transactions: ProgressiveTransactions::new(vec![]).unwrap(), + }, + signature: Signature::empty(), + }) + } + + /// Build the request items corresponding to validators 10, 11, and 12 + /// at inclusion list committee positions 0, 1 and 2. + fn request_items() -> InclusionListsByIndicesRequestItems { + InclusionListsByIndicesRequestItems::new(InclusionListsByIndicesRequestParams { + slot: SLOT, + dependent_root: DEPENDENT_ROOT, + requested: vec![(0, 10), (1, 11), (2, 12)], + }) + } + + #[test] + fn complete_response_resolves_request() { + let mut items = request_items(); + assert_eq!( + items.add(inclusion_list(SLOT, DEPENDENT_ROOT, 10)), + Ok(false) + ); + assert_eq!( + items.add(inclusion_list(SLOT, DEPENDENT_ROOT, 11)), + Ok(false) + ); + assert_eq!( + items.add(inclusion_list(SLOT, DEPENDENT_ROOT, 12)), + Ok(true) + ); + assert_eq!(items.consume().len(), 3); + } + + #[test] + fn rejects_lists_that_were_not_requested() { + let other_slot = SLOT + 1; + let other_dependent_root = Hash256::repeat_byte(2); + let unrequested_validator = 13; + + assert_eq!( + request_items().add(inclusion_list(other_slot, DEPENDENT_ROOT, 10)), + Err(LookupVerifyError::UnrequestedSlot(other_slot)) + ); + assert_eq!( + request_items().add(inclusion_list(SLOT, other_dependent_root, 10)), + Err(LookupVerifyError::UnrequestedDependentRoot( + other_dependent_root + )) + ); + assert_eq!( + request_items().add(inclusion_list(SLOT, DEPENDENT_ROOT, unrequested_validator)), + Err(LookupVerifyError::UnrequestedIndex(unrequested_validator)) + ); + } + + #[test] + fn rejects_a_second_list_from_the_same_validator() { + let mut items = request_items(); + assert_eq!( + items.add(inclusion_list(SLOT, DEPENDENT_ROOT, 10)), + Ok(false) + ); + assert_eq!( + items.add(inclusion_list(SLOT, DEPENDENT_ROOT, 10)), + Err(LookupVerifyError::DuplicatedData(SLOT, 10)) + ); + } + + #[test] + fn validator_holding_several_positions_completes_with_one_list() { + let mut items = + InclusionListsByIndicesRequestItems::new(InclusionListsByIndicesRequestParams { + slot: SLOT, + dependent_root: DEPENDENT_ROOT, + requested: vec![(0, 10), (1, 10), (2, 11)], + }); + assert_eq!( + items.add(inclusion_list(SLOT, DEPENDENT_ROOT, 10)), + Ok(false) + ); + assert_eq!( + items.add(inclusion_list(SLOT, DEPENDENT_ROOT, 11)), + Ok(true) + ); + } + + #[test] + fn try_into_request_sets_the_requested_positions() { + let params = InclusionListsByIndicesRequestParams { + slot: SLOT, + dependent_root: DEPENDENT_ROOT, + requested: vec![(0, 10), (5, 11)], + }; + let request = params.clone().try_into_request::().unwrap(); + assert_eq!(request.slot, SLOT); + assert_eq!(request.dependent_root, DEPENDENT_ROOT); + assert_eq!( + request + .indices + .iter() + .enumerate() + .filter(|(_, is_set)| *is_set) + .map(|(position, _)| position) + .collect::>(), + vec![0, 5] + ); + + // There are only 16 inclusion list committee positions + let out_of_range = InclusionListsByIndicesRequestParams { + requested: vec![(16, 10)], + ..params + }; + assert!(out_of_range.try_into_request::().is_err()); + } +} diff --git a/beacon_node/network/src/sync/tests/inclusion_lists.rs b/beacon_node/network/src/sync/tests/inclusion_lists.rs new file mode 100644 index 00000000000..4ec662d1756 --- /dev/null +++ b/beacon_node/network/src/sync/tests/inclusion_lists.rs @@ -0,0 +1,212 @@ +//! Tests for `InclusionListsByIndices` req/resp protocol + +use super::lookups::SimulateConfig; +use super::*; +use crate::sync::network_context::InclusionListsByIndicesRequestParams; +use eth2::types::ProposerPreparationData; +use types::{Address, Domain, EthSpec, InclusionList, SignedRoot}; + +impl TestRig { + /// A signed inclusion list for the current slot from the first committee member. + fn committee_inclusion_list(&self) -> Arc { + let chain = &self.harness.chain; + let slot = chain.slot().unwrap(); + let (committee, dependent_root) = chain + .inclusion_list_committee(chain.head_beacon_block_root(), slot) + .unwrap(); + let validator_index = committee[0]; + + let message = InclusionList { + slot, + validator_index, + dependent_root, + transactions: vec![vec![0xaa].try_into().unwrap()].try_into().unwrap(), + }; + let epoch = slot.epoch(E::slots_per_epoch()); + let domain = chain.spec.get_domain( + epoch, + Domain::InclusionListCommittee, + &chain.spec.fork_at_epoch(epoch), + chain.genesis_validators_root, + ); + let signature = self.harness.validator_keypairs[validator_index as usize] + .sk + .sign(message.signing_root(domain)); + Arc::new(SignedInclusionList { message, signature }) + } + + /// Makes peers serve `inclusion_list` at every committee position. + fn serve_inclusion_list(&mut self, inclusion_list: &Arc) { + self.network_inclusion_lists = + vec![inclusion_list.clone(); E::inclusion_list_committee_size()]; + } + + /// Requests the lists of the first `position_count` committee positions of the current slot + /// from a single peer. + fn request_inclusion_lists(&mut self, position_count: usize) { + self.new_connected_peer(); + let chain = &self.harness.chain; + let slot = chain.slot().unwrap(); + let (committee, dependent_root) = chain + .inclusion_list_committee(chain.head_beacon_block_root(), slot) + .unwrap(); + let params = InclusionListsByIndicesRequestParams { + slot, + dependent_root, + requested: committee + .iter() + .enumerate() + .take(position_count) + .map(|(position, validator_index)| (position, *validator_index)) + .collect(), + }; + self.sync_manager + .network_context() + .inclusion_lists_by_indices_request(params) + .unwrap(); + } + + fn stored_inclusion_lists( + &self, + inclusion_list: &SignedInclusionList, + ) -> Vec { + self.harness + .chain + .inclusion_list_store + .read() + .get_signed_inclusion_lists( + inclusion_list.message.slot, + inclusion_list.message.dependent_root, + &[inclusion_list.message.validator_index], + ) + } + + /// Registers the proposer of the next slot as one of this node's validators. + async fn register_next_slot_proposer(&self) { + let slot = self.harness.chain.slot().unwrap() + 1; + let proposer_index = self + .harness + .chain + .head_beacon_state_cloned() + .get_beacon_proposer_index(slot, &self.harness.spec) + .unwrap(); + + self.harness + .chain + .execution_layer + .as_ref() + .unwrap() + .update_proposer_preparation( + slot.epoch(E::slots_per_epoch()), + [( + &ProposerPreparationData { + validator_index: proposer_index as u64, + fee_recipient: Address::ZERO, + }, + &None, + )], + ) + .await; + } + + /// Asserts a single request was sent for every inclusion list committee position. + fn assert_all_inclusion_lists_requested(&mut self) { + let request = self + .pop_received_network_event(|event| match event { + NetworkMessage::SendRequest { + request: RequestType::InclusionListsByIndices(request), + .. + } => Some(request.clone()), + _ => None, + }) + .unwrap(); + + let chain = &self.harness.chain; + let slot = chain.slot().unwrap(); + let (committee, dependent_root) = chain + .inclusion_list_committee(chain.head_beacon_block_root(), slot) + .unwrap(); + let requested_positions = request + .indices + .iter() + .enumerate() + .filter(|(_, requested)| *requested) + .map(|(position, _)| position) + .collect::>(); + assert_eq!(request.slot, slot); + assert_eq!(request.dependent_root, dependent_root); + assert_eq!( + requested_positions, + (0..committee.len()).collect::>() + ); + self.assert_empty_network(); + } + + fn assert_inclusion_lists_by_indices_request_completed(&mut self) { + assert_eq!( + self.sync_manager + .network_context() + .inclusion_lists_by_indices_request_count(), + 0, + "inclusion lists by indices request should no longer be active" + ); + } +} + +#[tokio::test] +async fn inclusion_lists_by_indices_request_completes_and_imports() { + let mut r = TestRig::default(); + let inclusion_list = r.committee_inclusion_list(); + r.serve_inclusion_list(&inclusion_list); + r.request_inclusion_lists(3); + r.simulate(SimulateConfig::happy_path()).await; + r.assert_inclusion_lists_by_indices_request_completed(); + r.assert_no_penalties(); + assert_eq!( + r.stored_inclusion_lists(&inclusion_list), + vec![(*inclusion_list).clone()] + ); +} + +#[tokio::test] +async fn inclusion_lists_for_another_slot_penalize_peer() { + let mut r = TestRig::default(); + let inclusion_list = r.committee_inclusion_list(); + r.serve_inclusion_list(&inclusion_list); + r.request_inclusion_lists(3); + r.simulate(SimulateConfig::happy_path().with_wrong_inclusion_list_slot_n_times(1)) + .await; + r.assert_inclusion_lists_by_indices_request_completed(); + r.assert_penalties(&["UnrequestedSlot"]); + assert!(r.stored_inclusion_lists(&inclusion_list).is_empty()); +} + +#[tokio::test] +async fn inclusion_lists_by_indices_short_response_completes() { + let mut r = TestRig::default(); + r.request_inclusion_lists(3); + r.simulate(SimulateConfig::happy_path()).await; + r.assert_inclusion_lists_by_indices_request_completed(); + r.assert_no_penalties(); +} + +#[tokio::test] +async fn deadline_requests_missing_inclusion_lists() { + let Some(mut r) = TestRig::new_after_heze() else { + return; + }; + r.new_connected_peer(); + r.register_next_slot_proposer().await; + r.sync_manager.on_inclusion_list_deadline().await; + r.assert_all_inclusion_lists_requested(); +} + +#[tokio::test] +async fn deadline_without_our_proposer_sends_nothing() { + let Some(mut r) = TestRig::new_after_heze() else { + return; + }; + r.new_connected_peer(); + r.sync_manager.on_inclusion_list_deadline().await; + r.assert_empty_network(); +} diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs index a675fa0216d..17187f23aea 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -39,8 +39,8 @@ use tokio::sync::mpsc; use tracing::info; use types::{ BlobSidecar, BlockImportSource, ColumnIndex, DataColumnSidecar, DataColumnSubnetId, - ForkContext, ForkName, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, Slot, - Spec as E, + ForkContext, ForkName, Hash256, SignedBeaconBlock, SignedExecutionPayloadEnvelope, + SignedInclusionList, Slot, Spec as E, }; /// Extract the Gloas payload envelope (if any) carried by a stored `RangeSyncBlock`. @@ -103,6 +103,8 @@ pub struct SimulateConfig { ee_offline_for_n_range_responses: Option, /// Disconnect all peers after this many successful BlocksByRange responses. successful_range_responses_before_disconnect: Option, + /// Number of InclusionListsByIndices requests that return inclusion lists for another slot + return_wrong_inclusion_list_slot_n_times: usize, } impl SimulateConfig { @@ -227,6 +229,11 @@ impl SimulateConfig { self.successful_range_responses_before_disconnect = Some(n); self } + + pub(super) fn with_wrong_inclusion_list_slot_n_times(mut self, n: usize) -> Self { + self.return_wrong_inclusion_list_slot_n_times = n; + self + } } fn genesis_fork() -> ForkName { @@ -340,6 +347,7 @@ impl TestRig { fork_name, network_blocks_by_root: <_>::default(), network_blocks_by_slot: <_>::default(), + network_inclusion_lists: <_>::default(), penalties: <_>::default(), seen_lookups: <_>::default(), requests: <_>::default(), @@ -466,7 +474,8 @@ impl TestRig { } Work::RpcBlobs { process_fn } | Work::RpcCustodyColumn(process_fn) - | Work::RpcEnvelope(process_fn) => process_fn.await, + | Work::RpcEnvelope(process_fn) + | Work::RpcInclusionLists(process_fn) => process_fn.await, Work::ChainSegment { process_fn, process_id: (chain_id, batch_epoch), @@ -860,6 +869,36 @@ impl TestRig { self.send_rpc_envelopes_response(req_id, peer_id, &envelopes); } + (RequestType::InclusionListsByIndices(req), AppRequestId::Sync(req_id)) => { + // Serve one list per validator; a validator can hold several committee positions. + let mut served_validators = HashSet::new(); + let mut inclusion_lists = req + .indices + .iter() + .enumerate() + .filter(|(_, requested)| *requested) + .filter_map(|(position, _)| self.network_inclusion_lists.get(position).cloned()) + .filter(|inclusion_list| { + served_validators.insert(inclusion_list.message.validator_index) + }) + .collect::>(); + + // Return lists for another slot N times + if self + .complete_strategy + .return_wrong_inclusion_list_slot_n_times + > 0 + { + self.complete_strategy + .return_wrong_inclusion_list_slot_n_times -= 1; + for inclusion_list in &mut inclusion_lists { + Arc::make_mut(inclusion_list).message.slot += 1; + } + } + + self.send_rpc_inclusion_lists_response(req_id, peer_id, &inclusion_lists); + } + (RequestType::Status(_req), AppRequestId::Status) => { // Ignore Status requests for now } @@ -980,6 +1019,35 @@ impl TestRig { }); } + fn send_rpc_inclusion_lists_response( + &mut self, + sync_request_id: SyncRequestId, + peer_id: PeerId, + inclusion_lists: &[Arc], + ) { + let validators = inclusion_lists + .iter() + .map(|inclusion_list| inclusion_list.message.validator_index) + .collect::>(); + self.log(&format!( + "Completing request {sync_request_id:?} to {peer_id} with inclusion lists from validators {validators:?}" + )); + + for inclusion_list in inclusion_lists { + self.push_sync_message(SyncMessage::RpcInclusionList { + sync_request_id, + peer_id, + inclusion_list: Some(inclusion_list.clone()), + }); + } + // Stream termination + self.push_sync_message(SyncMessage::RpcInclusionList { + sync_request_id, + peer_id, + inclusion_list: None, + }); + } + fn send_rpc_envelopes_response( &mut self, sync_request_id: SyncRequestId, @@ -1647,6 +1715,10 @@ impl TestRig { genesis_fork().gloas_enabled().then(Self::default) } + pub(super) fn new_after_heze() -> Option { + genesis_fork().heze_enabled().then(Self::default) + } + pub fn new_fulu_peer_test(fulu_test_type: FuluTestType) -> Option { genesis_fork().fulu_enabled().then(|| { Self::new(TestRigConfig { diff --git a/beacon_node/network/src/sync/tests/mod.rs b/beacon_node/network/src/sync/tests/mod.rs index 5fdf51ef3a9..4deafa909de 100644 --- a/beacon_node/network/src/sync/tests/mod.rs +++ b/beacon_node/network/src/sync/tests/mod.rs @@ -21,8 +21,9 @@ use tokio::sync::mpsc; use tracing_subscriber::fmt::MakeWriter; use tracing_subscriber::layer::SubscriberExt; use tracing_subscriber::util::SubscriberInitExt; -use types::{ForkName, Hash256, Slot, Spec as E}; +use types::{ForkName, Hash256, SignedInclusionList, Slot, Spec as E}; +mod inclusion_lists; mod lookups; mod range; @@ -77,6 +78,8 @@ struct TestRig { /// Blocks that will be used in the test but may not be known to `harness` yet. network_blocks_by_root: HashMap>, network_blocks_by_slot: HashMap>, + /// Inclusion lists that will be served by peers, indexed by inclusion list committee position. + network_inclusion_lists: Vec>, penalties: Vec, /// All seen lookups through the test run seen_lookups: HashMap,