From 58be6a5406088a84b64193d403ab4a3afe196f53 Mon Sep 17 00:00:00 2001 From: conache Date: Fri, 2 Oct 2026 17:01:40 +0300 Subject: [PATCH 01/12] Register request type for inclusion lists by indices and setup chunk validation logic --- .../src/service/api_types.rs | 2 + beacon_node/network/src/sync/manager.rs | 15 +- .../src/sync/network_context/requests.rs | 2 + .../requests/inclusion_lists_by_indices.rs | 160 ++++++++++++++++++ 4 files changed, 178 insertions(+), 1 deletion(-) create mode 100644 beacon_node/network/src/sync/network_context/requests/inclusion_lists_by_indices.rs diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs index db2a36d80f0..5038e9053d7 100644 --- a/beacon_node/lighthouse_network/src/service/api_types.rs +++ b/beacon_node/lighthouse_network/src/service/api_types.rs @@ -33,6 +33,8 @@ pub enum SyncRequestId { DataColumnsByRange(DataColumnsByRangeRequestId), /// Payload envelopes by range request PayloadEnvelopesByRange(PayloadEnvelopesByRangeRequestId), + /// Inclusion Lists by indices request + InclusionListsByIndices { id: Id }, } /// Request ID for data_columns_by_root requests. Block lookups do not issue this request directly. diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index dae965a6d16..15c2e79556e 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -73,7 +73,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 @@ -520,6 +520,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)) + } } } @@ -1491,4 +1494,14 @@ impl SyncManager { } } } + + /// Handles receiving a response for an inclusion lists by indices request. + fn on_inclusion_lists_by_indices_response( + &mut self, + request_id: Id, + peer_id: PeerId, + inclusion_list: RpcEvent>, + ) { + todo!() + } } diff --git a/beacon_node/network/src/sync/network_context/requests.rs b/beacon_node/network/src/sync/network_context/requests.rs index 18e46be07e7..60733b18fb1 100644 --- a/beacon_node/network/src/sync/network_context/requests.rs +++ b/beacon_node/network/src/sync/network_context/requests.rs @@ -26,6 +26,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 +35,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..462cb8cb925 --- /dev/null +++ b/beacon_node/network/src/sync/network_context/requests/inclusion_lists_by_indices.rs @@ -0,0 +1,160 @@ +use lighthouse_network::rpc::methods::InclusionListsByIndicesRequest; +use std::sync::Arc; +use types::{EthSpec, SignedInclusionList}; + +use super::{ActiveRequestItems, LookupVerifyError}; + +pub struct InclusionListsByIndicesRequestItems { + request: InclusionListsByIndicesRequest, + /// Validator indices at the requested committee positions. + validator_indices: Vec, + items: Vec>, +} + +impl InclusionListsByIndicesRequestItems { + pub fn new( + request: InclusionListsByIndicesRequest, + requested_validator_indices: Vec, + ) -> Self { + Self { + request, + validator_indices: requested_validator_indices, + 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.validator_indices.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.validator_indices.len()) + } + + fn consume(&mut self) -> Vec { + std::mem::take(&mut self.items) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use bls::Signature; + use types::{ + Hash256, InclusionList, InclusionListBits, MinimalEthSpec as E, ProgressiveTransactions, + Slot, + }; + + 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 { + let mut indices = InclusionListBits::::new(); + for position in 0..3 { + indices.set(position, true).unwrap(); + } + let request = InclusionListsByIndicesRequest { + slot: SLOT, + dependent_root: DEPENDENT_ROOT, + indices, + }; + InclusionListsByIndicesRequestItems::new(request, vec![10, 11, 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)) + ); + } +} From fe710d9fd44f7d642b1cde6556644e6110b1e4b3 Mon Sep 17 00:00:00 2001 From: conache Date: Fri, 2 Oct 2026 18:36:17 +0300 Subject: [PATCH 02/12] Add send request handler --- .../src/service/api_types.rs | 7 +- beacon_node/network/src/metrics.rs | 8 ++ beacon_node/network/src/sync/manager.rs | 6 +- .../network/src/sync/network_context.rs | 68 +++++++++++- .../src/sync/network_context/requests.rs | 3 + .../requests/inclusion_lists_by_indices.rs | 101 +++++++++++++----- 6 files changed, 158 insertions(+), 35 deletions(-) diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs index 5038e9053d7..4566f007de5 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, @@ -34,7 +37,9 @@ pub enum SyncRequestId { /// Payload envelopes by range request PayloadEnvelopesByRange(PayloadEnvelopesByRangeRequestId), /// Inclusion Lists by indices request - InclusionListsByIndices { id: Id }, + 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 d2b1c411db5..94f2cf38671 100644 --- a/beacon_node/network/src/metrics.rs +++ b/beacon_node/network/src/metrics.rs @@ -507,6 +507,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/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index 15c2e79556e..861e58d2f93 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -57,8 +57,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}; @@ -1498,7 +1498,7 @@ impl SyncManager { /// Handles receiving a response for an inclusion lists by indices request. fn on_inclusion_lists_by_indices_response( &mut self, - request_id: Id, + request_id: InclusionListsByIndicesRequestId, peer_id: PeerId, inclusion_list: RpcEvent>, ) { diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index d4a27c21017..32bc45418f2 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, + 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; @@ -255,6 +257,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 +343,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 +374,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 +414,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 +425,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 +1237,53 @@ impl SyncNetworkContext { Ok(id) } + pub fn send_inclusion_lists_by_indices_request( + &mut self, + peer_id: PeerId, + request: InclusionListsByIndicesRequestParams, + request_span: Span, + ) -> Result { + let id = self.next_id(); + + 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, + "Inclusion lists by indices RPC request sent" + ); + + self.inclusion_lists_by_indices_requests.insert( + id, + peer_id, + // false = do not enforce that every requested list is returned. 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 } diff --git a/beacon_node/network/src/sync/network_context/requests.rs b/beacon_node/network/src/sync/network_context/requests.rs index 60733b18fb1..c962e8ac5fd 100644 --- a/beacon_node/network/src/sync/network_context/requests.rs +++ b/beacon_node/network/src/sync/network_context/requests.rs @@ -12,6 +12,9 @@ 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::{ + InclusionListsByIndicesRequestItems, InclusionListsByIndicesRequestParams, +}; pub use payload_envelopes_by_range::PayloadEnvelopesByRangeRequestItems; pub use payload_envelopes_by_root::{ PayloadEnvelopesByRootRequestItems, PayloadEnvelopesByRootSingleRequest, 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 index 462cb8cb925..8a34fab2fe0 100644 --- 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 @@ -1,30 +1,51 @@ use lighthouse_network::rpc::methods::InclusionListsByIndicesRequest; use std::sync::Arc; -use types::{EthSpec, SignedInclusionList}; +use types::{EthSpec, Hash256, InclusionListBits, SignedInclusionList, Slot}; use super::{ActiveRequestItems, LookupVerifyError}; -pub struct InclusionListsByIndicesRequestItems { - request: InclusionListsByIndicesRequest, - /// Validator indices at the requested committee positions. - validator_indices: Vec, +#[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: Vec<(usize, u64)>, +} + +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, items: Vec>, } -impl InclusionListsByIndicesRequestItems { - pub fn new( - request: InclusionListsByIndicesRequest, - requested_validator_indices: Vec, - ) -> Self { +impl InclusionListsByIndicesRequestItems { + pub fn new(request: InclusionListsByIndicesRequestParams) -> Self { Self { request, - validator_indices: requested_validator_indices, items: vec![], } } } -impl ActiveRequestItems for InclusionListsByIndicesRequestItems { +impl ActiveRequestItems for InclusionListsByIndicesRequestItems { type Item = Arc; /// Appends a chunk to this multi-item request. Returns `true` once an inclusion list has been @@ -41,7 +62,12 @@ impl ActiveRequestItems for InclusionListsByIndicesRequestItems { } let validator_index = inclusion_list.message.validator_index; - if !self.validator_indices.contains(&validator_index) { + if !self + .request + .requested + .iter() + .any(|(_, requested)| *requested == validator_index) + { return Err(LookupVerifyError::UnrequestedIndex(validator_index)); } @@ -54,7 +80,7 @@ impl ActiveRequestItems for InclusionListsByIndicesRequestItems { } self.items.push(inclusion_list); - Ok(self.items.len() >= self.validator_indices.len()) + Ok(self.items.len() >= self.request.requested.len()) } fn consume(&mut self) -> Vec { @@ -66,10 +92,7 @@ impl ActiveRequestItems for InclusionListsByIndicesRequestItems { mod tests { use super::*; use bls::Signature; - use types::{ - Hash256, InclusionList, InclusionListBits, MinimalEthSpec as E, ProgressiveTransactions, - Slot, - }; + use types::{InclusionList, MinimalEthSpec as E, ProgressiveTransactions}; const SLOT: Slot = Slot::new(1); const DEPENDENT_ROOT: Hash256 = Hash256::repeat_byte(1); @@ -92,17 +115,12 @@ mod tests { /// Build the request items corresponding to validators 10, 11, and 12 /// at inclusion list committee positions 0, 1 and 2. - fn request_items() -> InclusionListsByIndicesRequestItems { - let mut indices = InclusionListBits::::new(); - for position in 0..3 { - indices.set(position, true).unwrap(); - } - let request = InclusionListsByIndicesRequest { + fn request_items() -> InclusionListsByIndicesRequestItems { + InclusionListsByIndicesRequestItems::new(InclusionListsByIndicesRequestParams { slot: SLOT, dependent_root: DEPENDENT_ROOT, - indices, - }; - InclusionListsByIndicesRequestItems::new(request, vec![10, 11, 12]) + requested: vec![(0, 10), (1, 11), (2, 12)], + }) } #[test] @@ -157,4 +175,33 @@ mod tests { Err(LookupVerifyError::DuplicatedData(SLOT, 10)) ); } + + #[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()); + } } From 1a3e97262efe6ae7df886d4bb8064661b9008cba Mon Sep 17 00:00:00 2001 From: conache Date: Fri, 2 Oct 2026 19:46:42 +0300 Subject: [PATCH 03/12] Add response handling and roundtrip testing --- beacon_node/network/src/router.rs | 35 +++++++- beacon_node/network/src/sync/manager.rs | 42 +++++++++- .../network/src/sync/network_context.rs | 20 ++++- .../network/src/sync/tests/inclusion_lists.rs | 82 +++++++++++++++++++ beacon_node/network/src/sync/tests/lookups.rs | 64 ++++++++++++++- beacon_node/network/src/sync/tests/mod.rs | 5 +- 6 files changed, 240 insertions(+), 8 deletions(-) create mode 100644 beacon_node/network/src/sync/tests/inclusion_lists.rs diff --git a/beacon_node/network/src/router.rs b/beacon_node/network/src/router.rs index 66f08fe06da..882d6bfab30 100644 --- a/beacon_node/network/src/router.rs +++ b/beacon_node/network/src/router.rs @@ -26,7 +26,7 @@ use tokio_stream::wrappers::UnboundedReceiverStream; use tracing::{debug, error, trace, warn}; use types::{ BlobSidecar, DataColumnSidecar, EthSpec, ForkContext, PartialDataColumn, SignedBeaconBlock, - SignedExecutionPayloadEnvelope, + SignedExecutionPayloadEnvelope, SignedInclusionList, }; /// Handles messages from the network and routes them to the appropriate service to be handled. @@ -364,14 +364,18 @@ impl Router { 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(_) @@ -864,6 +868,29 @@ impl Router { } } } + + /// Handle an `InclusionListsByIndices` response from the peer. + pub 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"); + } + } } /// Wraps a Network Channel to employ various RPC related network functionality for the diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index 861e58d2f93..8e114f16034 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -136,6 +136,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), @@ -857,6 +864,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(); @@ -1256,6 +1268,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, @@ -1502,6 +1533,15 @@ impl SyncManager { peer_id: PeerId, inclusion_list: RpcEvent>, ) { - todo!() + if let Some(Ok(inclusion_lists)) = + self.network + .on_inclusion_lists_by_indices_response(request_id, peer_id, inclusion_list) + { + debug!( + %peer_id, + count = inclusion_lists.len(), + "Received inclusion lists by indices" + ); + } } } diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 32bc45418f2..f9b2dec4322 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -58,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; @@ -1512,6 +1513,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, @@ -1760,6 +1773,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, 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..1318eeee299 --- /dev/null +++ b/beacon_node/network/src/sync/tests/inclusion_lists.rs @@ -0,0 +1,82 @@ +//! Tests for `InclusionListsByIndices` req/resp protocol + +use super::lookups::SimulateConfig; +use super::*; +use crate::sync::network_context::InclusionListsByIndicesRequestParams; +use bls::Signature; +use tracing::Span; +use types::{InclusionList, ProgressiveTransactions}; + +const INCLUSION_LIST_SLOT: Slot = Slot::new(1); +const DEPENDENT_ROOT: Hash256 = Hash256::repeat_byte(1); + +impl TestRig { + /// Gives peers an inclusion list for each of the first `held` committee positions, then + /// requests the first `requested` positions from a single peer. + fn setup_inclusion_lists_by_indices_request(&mut self, requested: u64, held: u64) { + let peer_id = self.new_connected_peer(); + self.network_inclusion_lists = (0..held) + .map(|validator_index| { + Arc::new(SignedInclusionList { + message: InclusionList { + slot: INCLUSION_LIST_SLOT, + validator_index, + dependent_root: DEPENDENT_ROOT, + transactions: ProgressiveTransactions::new(vec![]).unwrap(), + }, + signature: Signature::empty(), + }) + }) + .collect(); + + let params = InclusionListsByIndicesRequestParams { + slot: INCLUSION_LIST_SLOT, + dependent_root: DEPENDENT_ROOT, + requested: (0..requested) + .map(|validator_index| (validator_index as usize, validator_index)) + .collect(), + }; + self.sync_manager + .network_context() + .send_inclusion_lists_by_indices_request(peer_id, params, Span::none()) + .unwrap(); + } + + 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() { + let mut r = TestRig::default(); + r.setup_inclusion_lists_by_indices_request(3, 3); + r.simulate(SimulateConfig::happy_path()).await; + r.assert_inclusion_lists_by_indices_request_completed(); + r.assert_no_penalties(); +} + +#[tokio::test] +async fn inclusion_lists_for_another_slot_penalize_peer() { + let mut r = TestRig::default(); + r.setup_inclusion_lists_by_indices_request(3, 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"]); +} + +#[tokio::test] +async fn inclusion_lists_by_indices_short_response_completes() { + let mut r = TestRig::default(); + r.setup_inclusion_lists_by_indices_request(3, 2); + r.simulate(SimulateConfig::happy_path()).await; + r.assert_inclusion_lists_by_indices_request_completed(); + r.assert_no_penalties(); +} diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs index d800faca67a..5142f5d5c83 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -40,7 +40,7 @@ use tracing::info; use types::{ BlobSidecar, BlockImportSource, ColumnIndex, DataColumnSidecar, DataColumnSubnetId, ForkContext, ForkName, Hash256, MinimalEthSpec as E, SignedBeaconBlock, - SignedExecutionPayloadEnvelope, Slot, + SignedExecutionPayloadEnvelope, SignedInclusionList, Slot, }; /// 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(), @@ -860,6 +868,31 @@ impl TestRig { self.send_rpc_envelopes_response(req_id, peer_id, &envelopes); } + (RequestType::InclusionListsByIndices(req), AppRequestId::Sync(req_id)) => { + let mut inclusion_lists = req + .indices + .iter() + .enumerate() + .filter(|(_, requested)| *requested) + .filter_map(|(position, _)| self.network_inclusion_lists.get(position).cloned()) + .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::Router) => { // Ignore Status requests for now } @@ -980,6 +1013,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, diff --git a/beacon_node/network/src/sync/tests/mod.rs b/beacon_node/network/src/sync/tests/mod.rs index 4e185cc0817..16e71baaa2a 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, MinimalEthSpec as E, Slot}; +use types::{ForkName, Hash256, MinimalEthSpec as E, SignedInclusionList, Slot}; +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, From 7f5aec778e6bdff027e8556c8c202923ce56d114 Mon Sep 17 00:00:00 2001 From: conache Date: Sat, 3 Oct 2026 17:12:23 +0300 Subject: [PATCH 04/12] Trigger requesting missing inclusion lists at the inclusion list deadline --- .../src/service/api_types.rs | 2 +- beacon_node/network/src/sync/manager.rs | 124 +++++++++++++++++- .../network/src/sync/network_context.rs | 37 +++++- .../src/sync/network_context/requests.rs | 3 +- .../requests/inclusion_lists_by_indices.rs | 43 ++++-- .../network/src/sync/tests/inclusion_lists.rs | 90 ++++++++++++- beacon_node/network/src/sync/tests/lookups.rs | 4 + 7 files changed, 283 insertions(+), 20 deletions(-) diff --git a/beacon_node/lighthouse_network/src/service/api_types.rs b/beacon_node/lighthouse_network/src/service/api_types.rs index 4566f007de5..f7c1704b63e 100644 --- a/beacon_node/lighthouse_network/src/service/api_types.rs +++ b/beacon_node/lighthouse_network/src/service/api_types.rs @@ -36,7 +36,7 @@ pub enum SyncRequestId { DataColumnsByRange(DataColumnsByRangeRequestId), /// Payload envelopes by range request PayloadEnvelopesByRange(PayloadEnvelopesByRangeRequestId), - /// Inclusion Lists by indices request + /// Inclusion lists by indices request InclusionListsByIndices { id: InclusionListsByIndicesRequestId, }, diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index 8e114f16034..e2bb4403a00 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -47,7 +47,10 @@ 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::{BeaconChain, BeaconChainTypes, EngineState}; use futures::StreamExt; @@ -795,6 +798,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! { @@ -819,10 +827,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<(Hash256, 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) => { diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index f9b2dec4322..6d69e831328 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -3,7 +3,7 @@ use self::custody::{ActiveCustodyRequest, Error as CustodyRequestError}; pub use self::requests::{ - BlocksByRootSingleRequest, DataColumnsByRootRequestParams, + BlocksByRootSingleRequest, DataColumnsByRootRequestParams, InclusionListCommitteePositions, InclusionListsByIndicesRequestParams, PayloadEnvelopesByRootSingleRequest, }; use super::SyncMessage; @@ -145,6 +145,7 @@ pub enum RpcRequestSendError { pub enum NoPeerError { BlockPeer, CustodyPeer(ColumnIndex), + InclusionListPeer, } #[derive(Debug, PartialEq, Eq)] @@ -1238,13 +1239,39 @@ impl SyncNetworkContext { Ok(id) } - pub fn send_inclusion_lists_by_indices_request( + /// Requests inclusion lists from a synced peer. + pub fn inclusion_lists_by_indices_request( &mut self, - peer_id: PeerId, request: InclusionListsByIndicesRequestParams, - request_span: Span, ) -> 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, @@ -1270,7 +1297,7 @@ impl SyncNetworkContext { requested = ?request.requested, peer = %peer_id, %id, - "Inclusion lists by indices RPC request sent" + "Sync RPC request sent" ); self.inclusion_lists_by_indices_requests.insert( diff --git a/beacon_node/network/src/sync/network_context/requests.rs b/beacon_node/network/src/sync/network_context/requests.rs index c962e8ac5fd..ade3d5cd48e 100644 --- a/beacon_node/network/src/sync/network_context/requests.rs +++ b/beacon_node/network/src/sync/network_context/requests.rs @@ -13,7 +13,8 @@ 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::{ - InclusionListsByIndicesRequestItems, InclusionListsByIndicesRequestParams, + InclusionListCommitteePositions, InclusionListsByIndicesRequestItems, + InclusionListsByIndicesRequestParams, }; pub use payload_envelopes_by_range::PayloadEnvelopesByRangeRequestItems; pub use payload_envelopes_by_root::{ 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 index 8a34fab2fe0..7160d05837c 100644 --- 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 @@ -1,16 +1,20 @@ 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: Vec<(usize, u64)>, + pub requested: InclusionListCommitteePositions, } impl InclusionListsByIndicesRequestParams { @@ -33,13 +37,22 @@ impl InclusionListsByIndicesRequestParams { 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![], } } @@ -62,12 +75,7 @@ impl ActiveRequestItems for InclusionListsByIndicesRequestItems { } let validator_index = inclusion_list.message.validator_index; - if !self - .request - .requested - .iter() - .any(|(_, requested)| *requested == validator_index) - { + if !self.requested_validators.contains(&validator_index) { return Err(LookupVerifyError::UnrequestedIndex(validator_index)); } @@ -80,7 +88,8 @@ impl ActiveRequestItems for InclusionListsByIndicesRequestItems { } self.items.push(inclusion_list); - Ok(self.items.len() >= self.request.requested.len()) + + Ok(self.items.len() >= self.requested_validators.len()) } fn consume(&mut self) -> Vec { @@ -176,6 +185,24 @@ mod tests { ); } + #[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 { diff --git a/beacon_node/network/src/sync/tests/inclusion_lists.rs b/beacon_node/network/src/sync/tests/inclusion_lists.rs index 1318eeee299..abf56f6e7c9 100644 --- a/beacon_node/network/src/sync/tests/inclusion_lists.rs +++ b/beacon_node/network/src/sync/tests/inclusion_lists.rs @@ -4,8 +4,8 @@ use super::lookups::SimulateConfig; use super::*; use crate::sync::network_context::InclusionListsByIndicesRequestParams; use bls::Signature; -use tracing::Span; -use types::{InclusionList, ProgressiveTransactions}; +use eth2::types::ProposerPreparationData; +use types::{Address, EthSpec, InclusionList, ProgressiveTransactions}; const INCLUSION_LIST_SLOT: Slot = Slot::new(1); const DEPENDENT_ROOT: Hash256 = Hash256::repeat_byte(1); @@ -14,7 +14,7 @@ impl TestRig { /// Gives peers an inclusion list for each of the first `held` committee positions, then /// requests the first `requested` positions from a single peer. fn setup_inclusion_lists_by_indices_request(&mut self, requested: u64, held: u64) { - let peer_id = self.new_connected_peer(); + self.new_connected_peer(); self.network_inclusion_lists = (0..held) .map(|validator_index| { Arc::new(SignedInclusionList { @@ -38,10 +38,71 @@ impl TestRig { }; self.sync_manager .network_context() - .send_inclusion_lists_by_indices_request(peer_id, params, Span::none()) + .inclusion_lists_by_indices_request(params) .unwrap(); } + /// 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 @@ -80,3 +141,24 @@ async fn inclusion_lists_by_indices_short_response_completes() { 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 5142f5d5c83..3d9d392818a 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -1709,6 +1709,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 { From bfd49ecee59c6da6eba763b964c7e7c1ab420e95 Mon Sep 17 00:00:00 2001 From: conache Date: Sat, 3 Oct 2026 17:46:41 +0300 Subject: [PATCH 05/12] Wire inclusion lists handling fetched using rpc --- beacon_node/beacon_processor/src/lib.rs | 10 ++++++++ .../src/scheduler/work_queue.rs | 6 +++++ .../src/network_beacon_processor/mod.rs | 16 +++++++++++++ .../network_beacon_processor/sync_methods.rs | 19 +++++++++++++++ beacon_node/network/src/sync/manager.rs | 20 +++++++++------- .../network/src/sync/network_context.rs | 23 +++++++++++++++++++ beacon_node/network/src/sync/tests/lookups.rs | 3 ++- 7 files changed, 88 insertions(+), 9 deletions(-) diff --git a/beacon_node/beacon_processor/src/lib.rs b/beacon_node/beacon_processor/src/lib.rs index c783c250642..a71b77f07ea 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, @@ -856,6 +859,8 @@ impl BeaconProcessor { Some(item) } else if let Some(item) = work_queues.rpc_envelope_queue.pop() { Some(item) + } else if let Some(item) = work_queues.rpc_inclusion_lists_queue.pop() { + Some(item) // Check delayed blocks before gossip blocks, the gossip blocks might rely // on the delayed ones. } else if let Some(item) = work_queues.delayed_block_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 56e7d049c9a..79699fa6d94 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,8 @@ impl BeaconProcessorQueueLengths { rpc_custody_column_queue: 64, // Bounded by `PARENT_DEPTH_TOLERANCE`; one envelope per Gloas block. rpc_envelope_queue: 1024, + // One request per slot, each returning at most the inclusion list committee size. + rpc_inclusion_lists_queue: 64, column_reconstruction_queue: 1, chain_segment_queue: 64, backfill_chain_segment: 64, @@ -269,6 +272,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 +349,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 +423,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/network/src/network_beacon_processor/mod.rs b/beacon_node/network/src/network_beacon_processor/mod.rs index 65beb0baa0f..7ba954b4e52 100644 --- a/beacon_node/network/src/network_beacon_processor/mod.rs +++ b/beacon_node/network/src/network_beacon_processor/mod.rs @@ -601,6 +601,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 f5549548c39..ba79a6d546a 100644 --- a/beacon_node/network/src/network_beacon_processor/sync_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/sync_methods.rs @@ -303,6 +303,25 @@ 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" + ); + } + /// 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/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index e2bb4403a00..06648e3c9e7 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -1655,15 +1655,19 @@ impl SyncManager { peer_id: PeerId, inclusion_list: RpcEvent>, ) { - if let Some(Ok(inclusion_lists)) = - self.network - .on_inclusion_lists_by_indices_response(request_id, peer_id, inclusion_list) + 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, - count = inclusion_lists.len(), - "Received inclusion lists by indices" - ); + 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 6d69e831328..01271f88b53 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -1675,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, diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs index 3d9d392818a..cf72220edfc 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -474,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), From 85828c90699a54f41c219a2bec49fab8344dd152 Mon Sep 17 00:00:00 2001 From: conache Date: Sat, 3 Oct 2026 18:50:39 +0300 Subject: [PATCH 06/12] Add rpc inclusion lists verification logic --- .../gossip_verified_inclusion_list.rs | 80 ++++++++++++++---- .../src/inclusion_list_verification/mod.rs | 33 ++++++++ .../src/inclusion_list_verification/tests.rs | 4 +- .../gossip_methods.rs | 8 +- .../network_beacon_processor/sync_methods.rs | 26 ++++++ .../src/network_beacon_processor/tests.rs | 83 +++++++++++++++++++ 6 files changed, 213 insertions(+), 21 deletions(-) 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 56f1704c586..6d60076d4ea 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; @@ -15,6 +15,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>, @@ -43,21 +44,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 @@ -200,9 +208,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, @@ -218,16 +261,18 @@ impl BeaconChain { pub fn verify_inclusion_list_for_gossip( &self, signed_inclusion_list: SignedInclusionList, + source: InclusionListSource, ) -> Result { 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, &ctx) { Ok(verified) => { debug!( %slot, %validator_index, + ?source, is_timely = verified.is_timely, "Successfully verified gossip inclusion list" ); @@ -241,6 +286,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 6c6e2a2ede4..20df91e13a4 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 079ce7471a2..9fbf8a09c46 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}, @@ -62,7 +62,7 @@ impl TestContext { fn gossip_ctx(&self) -> GossipVerificationContext<'_, T> { self.harness .chain - .inclusion_list_gossip_verification_context() + .inclusion_list_gossip_verification_context(InclusionListSource::Gossip) } fn current_slot(&self) -> Slot { 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 c1467ebcf75..a8c63d7e11a 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; 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::{ @@ -4368,7 +4370,9 @@ impl NetworkBeaconProcessor { peer_id: PeerId, inclusion_list: Box, ) { - let verification_result = self.chain.verify_inclusion_list_for_gossip(*inclusion_list); + let verification_result = self + .chain + .verify_inclusion_list_for_gossip(*inclusion_list, InclusionListSource::Gossip); match verification_result { Ok(verified_inclusion_list) => { 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 ba79a6d546a..6d8986ab9c2 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,6 +27,7 @@ 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 std::sync::Arc; @@ -320,6 +323,29 @@ impl NetworkBeaconProcessor { count = inclusion_lists.len(), "Processing RPC inclusion lists" ); + + 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), + 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", + }); + } + Err(_) => {} + } + } } /// Attempt to verify and import an execution payload envelope received via RPC. diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index ae2525dfe86..3c53ab2860b 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -3227,3 +3227,86 @@ async fn test_gossip_inclusion_list_propagation_follows_the_store_outcome() { } } } + +// 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(); + + assert!( + rig.receive_network_messages_with_timeout(Duration::from_millis(500), None) + .await + .is_none(), + "a valid list should not penalize the peer" + ); + assert_eq!( + rig.chain + .inclusion_list_store + .read() + .get_signed_inclusion_lists(slot, dependent_root, &[validator_index]), + vec![inclusion_list] + ); +} + +// 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() + ); +} From 774454c346635c3eb28160ff9a34447f3036ea2b Mon Sep 17 00:00:00 2001 From: conache Date: Sat, 3 Oct 2026 19:23:01 +0300 Subject: [PATCH 07/12] Add verification tests --- .../src/inclusion_list_verification/tests.rs | 60 +++++++++++++++++++ 1 file changed, 60 insertions(+) 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 9fbf8a09c46..fcb48d2581f 100644 --- a/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs +++ b/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs @@ -65,6 +65,12 @@ impl TestContext { .inclusion_list_gossip_verification_context(InclusionListSource::Gossip) } + fn rpc_ctx(&self) -> GossipVerificationContext<'_, T> { + self.harness + .chain + .inclusion_list_gossip_verification_context(InclusionListSource::Rpc) + } + fn current_slot(&self) -> Slot { self.harness.chain.slot().expect("should read slot") } @@ -378,3 +384,57 @@ async fn dependent_root_must_be_the_shuffling_dependent_block() { Err(InclusionListVerificationError::InvalidDependentRoot { .. }) )); } + +#[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.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.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]]); + + 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.gossip_ctx()), + Err(InclusionListVerificationError::PastSlot { .. }) + )); + assert!(GossipVerifiedInclusionList::new(signed.clone(), &ctx.rpc_ctx()).is_ok()); + + ctx.harness + .chain + .slot_clock + .set_slot(first_unretained_slot.as_u64()); + assert!(matches!( + GossipVerifiedInclusionList::new(signed, &ctx.rpc_ctx()), + Err(InclusionListVerificationError::PastSlot { .. }) + )); +} From 7db241744cb26153d81e8b9638e095c0dc0fff2d Mon Sep 17 00:00:00 2001 From: conache Date: Sat, 3 Oct 2026 21:34:17 +0300 Subject: [PATCH 08/12] Update roundtrip tests to assert the fetched inclusion lists --- .../network/src/sync/tests/inclusion_lists.rs | 107 +++++++++++++----- beacon_node/network/src/sync/tests/lookups.rs | 5 + 2 files changed, 82 insertions(+), 30 deletions(-) diff --git a/beacon_node/network/src/sync/tests/inclusion_lists.rs b/beacon_node/network/src/sync/tests/inclusion_lists.rs index abf56f6e7c9..ca98f37e4a2 100644 --- a/beacon_node/network/src/sync/tests/inclusion_lists.rs +++ b/beacon_node/network/src/sync/tests/inclusion_lists.rs @@ -3,37 +3,60 @@ use super::lookups::SimulateConfig; use super::*; use crate::sync::network_context::InclusionListsByIndicesRequestParams; -use bls::Signature; use eth2::types::ProposerPreparationData; -use types::{Address, EthSpec, InclusionList, ProgressiveTransactions}; - -const INCLUSION_LIST_SLOT: Slot = Slot::new(1); -const DEPENDENT_ROOT: Hash256 = Hash256::repeat_byte(1); +use types::{Address, Domain, EthSpec, InclusionList, SignedRoot}; impl TestRig { - /// Gives peers an inclusion list for each of the first `held` committee positions, then - /// requests the first `requested` positions from a single peer. - fn setup_inclusion_lists_by_indices_request(&mut self, requested: u64, held: u64) { - self.new_connected_peer(); - self.network_inclusion_lists = (0..held) - .map(|validator_index| { - Arc::new(SignedInclusionList { - message: InclusionList { - slot: INCLUSION_LIST_SLOT, - validator_index, - dependent_root: DEPENDENT_ROOT, - transactions: ProgressiveTransactions::new(vec![]).unwrap(), - }, - signature: Signature::empty(), - }) - }) - .collect(); + /// 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 first `positions` committee positions of the current slot from a single peer. + fn request_inclusion_lists(&mut self, positions: 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: INCLUSION_LIST_SLOT, - dependent_root: DEPENDENT_ROOT, - requested: (0..requested) - .map(|validator_index| (validator_index as usize, validator_index)) + slot, + dependent_root, + requested: committee + .iter() + .enumerate() + .take(positions) + .map(|(position, validator_index)| (position, *validator_index)) .collect(), }; self.sync_manager @@ -42,6 +65,21 @@ impl TestRig { .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; @@ -115,28 +153,37 @@ impl TestRig { } #[tokio::test] -async fn inclusion_lists_by_indices_request_completes() { +async fn inclusion_lists_by_indices_request_completes_and_imports() { let mut r = TestRig::default(); - r.setup_inclusion_lists_by_indices_request(3, 3); + 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(); - r.setup_inclusion_lists_by_indices_request(3, 3); + 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.setup_inclusion_lists_by_indices_request(3, 2); + r.request_inclusion_lists(3); r.simulate(SimulateConfig::happy_path()).await; r.assert_inclusion_lists_by_indices_request_completed(); r.assert_no_penalties(); diff --git a/beacon_node/network/src/sync/tests/lookups.rs b/beacon_node/network/src/sync/tests/lookups.rs index cf72220edfc..d5227a90021 100644 --- a/beacon_node/network/src/sync/tests/lookups.rs +++ b/beacon_node/network/src/sync/tests/lookups.rs @@ -870,12 +870,17 @@ impl TestRig { } (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 From 5c8a93527d541ee9e1bf459bad5d71091d62198a Mon Sep 17 00:00:00 2001 From: conache Date: Sat, 3 Oct 2026 21:35:49 +0300 Subject: [PATCH 09/12] Register inclusion_lists_by_indices metrics --- beacon_node/network/src/sync/network_context.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 01271f88b53..54f8f792840 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -1861,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); } From 03da5a51be0d11e85db84043de9042d7857811dc Mon Sep 17 00:00:00 2001 From: conache Date: Sat, 3 Oct 2026 21:48:05 +0300 Subject: [PATCH 10/12] Cleanup fixes --- beacon_node/beacon_processor/src/lib.rs | 4 +-- .../src/network_beacon_processor/tests.rs | 26 +++++++++++++------ beacon_node/network/src/sync/manager.rs | 3 ++- 3 files changed, 22 insertions(+), 11 deletions(-) diff --git a/beacon_node/beacon_processor/src/lib.rs b/beacon_node/beacon_processor/src/lib.rs index a71b77f07ea..44d6f056357 100644 --- a/beacon_node/beacon_processor/src/lib.rs +++ b/beacon_node/beacon_processor/src/lib.rs @@ -859,8 +859,6 @@ impl BeaconProcessor { Some(item) } else if let Some(item) = work_queues.rpc_envelope_queue.pop() { Some(item) - } else if let Some(item) = work_queues.rpc_inclusion_lists_queue.pop() { - Some(item) // Check delayed blocks before gossip blocks, the gossip blocks might rely // on the delayed ones. } else if let Some(item) = work_queues.delayed_block_queue.pop() { @@ -1009,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() { diff --git a/beacon_node/network/src/network_beacon_processor/tests.rs b/beacon_node/network/src/network_beacon_processor/tests.rs index 3c53ab2860b..292b0870c9e 100644 --- a/beacon_node/network/src/network_beacon_processor/tests.rs +++ b/beacon_node/network/src/network_beacon_processor/tests.rs @@ -3261,19 +3261,29 @@ async fn test_rpc_inclusion_list_is_imported() { .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!( - rig.receive_network_messages_with_timeout(Duration::from_millis(500), None) + 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" ); - assert_eq!( - rig.chain - .inclusion_list_store - .read() - .get_signed_inclusion_lists(slot, dependent_root, &[validator_index]), - vec![inclusion_list] - ); } // A list fetched over RPC from a validator outside the committee diff --git a/beacon_node/network/src/sync/manager.rs b/beacon_node/network/src/sync/manager.rs index 06648e3c9e7..650d2569d3e 100644 --- a/beacon_node/network/src/sync/manager.rs +++ b/beacon_node/network/src/sync/manager.rs @@ -52,6 +52,7 @@ use crate::sync::network_context::{ 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; @@ -919,7 +920,7 @@ impl SyncManager { fn get_missing_inclusion_lists( &self, slot: Slot, - ) -> Option<(Hash256, InclusionListCommitteePositions)> { + ) -> Option<(DependentRoot, InclusionListCommitteePositions)> { let head_block_root = self.chain.canonical_head.cached_head().head_block_root(); let (committee, dependent_root) = self From 6f876dc9b227af074a9410f853dca766ba2f2ed3 Mon Sep 17 00:00:00 2001 From: conache Date: Mon, 5 Oct 2026 18:52:43 +0300 Subject: [PATCH 11/12] List for future slot check --- .../beacon_chain/src/inclusion_list_verification/tests.rs | 7 +++++++ beacon_node/beacon_processor/src/scheduler/work_queue.rs | 1 - .../network/src/network_beacon_processor/sync_methods.rs | 2 ++ 3 files changed, 9 insertions(+), 1 deletion(-) 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 fcb48d2581f..4fe18194853 100644 --- a/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs +++ b/beacon_node/beacon_chain/src/inclusion_list_verification/tests.rs @@ -413,6 +413,13 @@ fn rpc_list_within_retention_window() { 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.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; diff --git a/beacon_node/beacon_processor/src/scheduler/work_queue.rs b/beacon_node/beacon_processor/src/scheduler/work_queue.rs index 79699fa6d94..2652d656df7 100644 --- a/beacon_node/beacon_processor/src/scheduler/work_queue.rs +++ b/beacon_node/beacon_processor/src/scheduler/work_queue.rs @@ -207,7 +207,6 @@ impl BeaconProcessorQueueLengths { rpc_custody_column_queue: 64, // Bounded by `PARENT_DEPTH_TOLERANCE`; one envelope per Gloas block. rpc_envelope_queue: 1024, - // One request per slot, each returning at most the inclusion list committee size. rpc_inclusion_lists_queue: 64, column_reconstruction_queue: 1, chain_segment_queue: 64, 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 6d8986ab9c2..5504fd69568 100644 --- a/beacon_node/network/src/network_beacon_processor/sync_methods.rs +++ b/beacon_node/network/src/network_beacon_processor/sync_methods.rs @@ -343,6 +343,8 @@ impl NetworkBeaconProcessor { 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(_) => {} } } From 962911d933c946b45238d4abc5686ca526f176d3 Mon Sep 17 00:00:00 2001 From: conache Date: Tue, 6 Oct 2026 12:51:45 +0300 Subject: [PATCH 12/12] Cleanup: use better namings --- beacon_node/network/src/sync/network_context.rs | 4 ++-- beacon_node/network/src/sync/tests/inclusion_lists.rs | 7 ++++--- 2 files changed, 6 insertions(+), 5 deletions(-) diff --git a/beacon_node/network/src/sync/network_context.rs b/beacon_node/network/src/sync/network_context.rs index 54f8f792840..7cb2989fa4a 100644 --- a/beacon_node/network/src/sync/network_context.rs +++ b/beacon_node/network/src/sync/network_context.rs @@ -1303,8 +1303,8 @@ impl SyncNetworkContext { self.inclusion_lists_by_indices_requests.insert( id, peer_id, - // false = do not enforce that every requested list is returned. A peer may hold fewer - // inclusion lists than requested. + // 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, diff --git a/beacon_node/network/src/sync/tests/inclusion_lists.rs b/beacon_node/network/src/sync/tests/inclusion_lists.rs index ca98f37e4a2..4ec662d1756 100644 --- a/beacon_node/network/src/sync/tests/inclusion_lists.rs +++ b/beacon_node/network/src/sync/tests/inclusion_lists.rs @@ -41,8 +41,9 @@ impl TestRig { vec![inclusion_list.clone(); E::inclusion_list_committee_size()]; } - /// Requests the first `positions` committee positions of the current slot from a single peer. - fn request_inclusion_lists(&mut self, positions: usize) { + /// 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(); @@ -55,7 +56,7 @@ impl TestRig { requested: committee .iter() .enumerate() - .take(positions) + .take(position_count) .map(|(position, validator_index)| (position, *validator_index)) .collect(), };