feat: network sync refactor - min req message size

beacon_node/execution_layer/src/eip8025/state.rs

@@ -51,13 +51,17 @@ impl State {
         Self::default()
     }
 
-    /// Return buffer entries that do not yet have sufficient proofs for promotion,
-    /// restricted to those on the ancestor path required to satisfy `latest_fcs`.
+    /// Return all buffer entries on the ancestor path required to satisfy `latest_fcs`,
+    /// including entries that already have sufficient proofs.
+    ///
+    /// Complete entries are returned so the sync layer can include them as skip-filters in
+    /// `ExecutionProofsByRange` requests, telling the serving peer not to re-send proofs
+    /// for blocks the requester already holds. Callers should inspect `existing_proof_types`
+    /// against the configured proof type set to determine which entries are still missing.
     ///
     /// If `latest_fcs` is unset there is no pending fork-choice update to satisfy, so
     /// nothing is returned. Otherwise the buffer is walked backwards from
-    /// `latest_fcs.head_block_hash`; entries that lack sufficient proofs are collected
-    /// until a block is not found in the buffer (reached the tree or an unseen block).
+    /// `latest_fcs.head_block_hash` until a block is not found in the buffer.
     pub fn missing_proofs(&self) -> Vec<MissingProofInfo> {
         let Some(latest_fcs) = &self.latest_fcs else {
             return vec![];
@@ -71,22 +75,20 @@ impl State {
             .map(|p| (p.metadata.block_hash, p))
             .collect();
 
-        // Walk backwards from the FCS head through buffer entries, collecting
-        // those that still lack sufficient proofs. Stop when a block is not in
-        // the buffer (reached the tree or an unseen block).
+        // Walk backwards from the FCS head through buffer entries, collecting all
+        // entries (missing and complete). Stop when a block is not in the buffer
+        // (reached the tree or an unseen block).
         let mut result = Vec::new();
         let mut current = latest_fcs.head_block_hash;
         loop {
             let Some(req) = buffer_by_block_hash.get(&current) else {
                 break;
             };
-            if req.proofs.len() < self.min_required_proofs {
-                result.push(MissingProofInfo {
-                    root: req.metadata.request_root,
-                    existing_proof_types: req.proofs.iter().map(|p| p.message.proof_type).collect(),
-                    slot: Default::default(), // populated by BeaconChain::missing_execution_proofs()
-                });
-            }
+            result.push(MissingProofInfo {
+                root: req.metadata.request_root,
+                existing_proof_types: req.proofs.iter().map(|p| p.message.proof_type).collect(),
+                slot: Default::default(), // populated by BeaconChain::missing_execution_proofs()
+            });
             current = req.metadata.parent_hash;
         }
 

beacon_node/lighthouse_network/src/config.rs

@@ -22,7 +22,6 @@ pub const DEFAULT_TCP_PORT: u16 = 9000u16;
 pub const DEFAULT_DISC_PORT: u16 = 9000u16;
 pub const DEFAULT_QUIC_PORT: u16 = 9001u16;
 pub const DEFAULT_IDONTWANT_MESSAGE_SIZE_THRESHOLD: usize = 1000usize;
-pub const DEFAULT_PROOF_SYNC_ACTIVATION_SLOTS: u64 = 10;
 
 pub struct GossipsubConfigParams {
     pub message_domain_valid_snappy: [u8; 4],
@@ -133,11 +132,6 @@ pub struct Config {
     /// Proof types supported by this client.
     pub proof_types: Option<Vec<u8>>,
 
-    /// Number of slot ticks to wait after range sync completes before issuing
-    /// `ExecutionProofsByRange` requests. Gives the beacon processor time to finish
-    /// calling `notify_new_payload` for all imported blocks before proofs are requested.
-    pub proof_sync_activation_slots: u64,
-
     /// Configuration for the outbound rate limiter (requests made by this node).
     pub outbound_rate_limiter_config: Option<OutboundRateLimiterConfig>,
 
@@ -374,7 +368,6 @@ impl Default for Config {
             enable_light_client_server: true,
             enable_execution_proof: false,
             proof_types: None,
-            proof_sync_activation_slots: DEFAULT_PROOF_SYNC_ACTIVATION_SLOTS,
             outbound_rate_limiter_config: None,
             invalid_block_storage: None,
             inbound_rate_limiter_config: None,

beacon_node/lighthouse_network/src/rpc/codec.rs

@@ -595,8 +595,12 @@ fn handle_rpc_request<E: EthSpec>(
             )))
         }
         SupportedProtocol::ExecutionProofsByRangeV1 => {
+            let max_filters = spec.max_request_blocks(current_fork);
             Ok(Some(RequestType::ExecutionProofsByRange(
-                ExecutionProofsByRangeRequest::from_ssz_bytes(decoded_buffer)?,
+                ExecutionProofsByRangeRequest::from_ssz_bytes_with_max(
+                    decoded_buffer,
+                    max_filters,
+                )?,
             )))
         }
         SupportedProtocol::ExecutionProofsByRootV1 => Ok(Some(RequestType::ExecutionProofsByRoot(

beacon_node/lighthouse_network/src/rpc/methods.rs

@@ -585,12 +585,22 @@ pub struct ExecutionProofStatus {
 }
 
 /// Request execution proofs for a slot range from a peer.
-#[derive(Encode, Decode, Clone, Debug, PartialEq)]
+///
+/// `proof_filters` is an optional per-block filter that tells the peer which proof types we still
+/// need for specific blocks in the range. Blocks not listed in `proof_filters` will have all known
+/// proof types returned; blocks listed will only have the specified types returned. This avoids
+/// transferring proof types the requester already holds.
+///
+/// Matches the `ExecutionProofsByRange` request type in the EIP-8025 p2p spec.
+#[derive(Clone, Debug, PartialEq)]
 pub struct ExecutionProofsByRangeRequest {
     /// The starting slot to request execution proofs.
     pub start_slot: u64,
     /// The number of slots from the start slot.
     pub count: u64,
+    /// Per-block proof-type filters for blocks in the range where only some proof types are needed.
+    /// Empty list means "return all proof types for every block in the range."
+    pub proof_filters: RuntimeVariableList<ProofByRootIdentifier>,
 }
 
 impl ExecutionProofsByRangeRequest {
@@ -601,26 +611,76 @@ impl ExecutionProofsByRangeRequest {
             .saturating_mul(MaxExecutionProofsPerPayload::to_u64())
     }
 
+    /// Minimum SSZ encoded byte length: the two fixed `u64` fields plus the 4-byte offset pointer
+    /// for the variable-length `proof_filters` list.
     pub fn ssz_min_len() -> usize {
-        ExecutionProofsByRangeRequest {
-            start_slot: 0,
-            count: 0,
-        }
-        .as_ssz_bytes()
-        .len()
+        // start_slot (8) + count (8) + proof_filters offset (4)
+        20
     }
 
-    pub fn ssz_max_len() -> usize {
-        Self::ssz_min_len()
+    /// Maximum SSZ encoded byte length when `proof_filters` holds up to `max_request_blocks`
+    /// entries.
+    ///
+    /// Each `ProofByRootIdentifier` is a variable-length SSZ container:
+    /// - `block_root`: 32 bytes (fixed)
+    /// - `proof_types` offset field: 4 bytes (within the container fixed portion)
+    /// - `proof_types` content: at most `MAX_EXECUTION_PROOFS_PER_PAYLOAD` × 1 byte = 4 bytes
+    ///
+    /// A `List` of `max_request_blocks` variable-length items also requires a 4-byte offset table
+    /// entry per item.
+    pub fn ssz_max_len(max_request_blocks: usize) -> usize {
+        const MAX_PROOF_BY_ROOT_IDENTIFIER_BYTES: usize = 32 + 4 + 4;
+        Self::ssz_min_len() + max_request_blocks * (4 + MAX_PROOF_BY_ROOT_IDENTIFIER_BYTES)
+    }
+
+    /// Decode from SSZ bytes, supplying a runtime maximum for the `proof_filters` list length.
+    pub fn from_ssz_bytes_with_max(
+        bytes: &[u8],
+        max_filters: usize,
+    ) -> Result<Self, ssz::DecodeError> {
+        let mut builder = ssz::SszDecoderBuilder::new(bytes);
+        builder.register_type::<u64>()?;
+        builder.register_type::<u64>()?;
+        builder.register_anonymous_variable_length_item()?;
+        let mut decoder = builder.build()?;
+        Ok(Self {
+            start_slot: decoder.decode_next::<u64>()?,
+            count: decoder.decode_next::<u64>()?,
+            proof_filters: decoder.decode_next_with(|slice| {
+                RuntimeVariableList::from_ssz_bytes(slice, max_filters)
+            })?,
+        })
+    }
+}
+
+impl ssz::Encode for ExecutionProofsByRangeRequest {
+    fn is_ssz_fixed_len() -> bool {
+        false
+    }
+
+    fn ssz_append(&self, buf: &mut Vec<u8>) {
+        // Fixed portion: start_slot (8) + count (8) + proof_filters offset (4) = 20 bytes.
+        let num_fixed_bytes = 8 + 8 + ssz::BYTES_PER_LENGTH_OFFSET;
+        let mut encoder = ssz::SszEncoder::container(buf, num_fixed_bytes);
+        encoder.append(&self.start_slot);
+        encoder.append(&self.count);
+        encoder.append(&self.proof_filters);
+        encoder.finalize();
+    }
+
+    fn ssz_bytes_len(&self) -> usize {
+        8 + 8 + ssz::BYTES_PER_LENGTH_OFFSET + self.proof_filters.ssz_bytes_len()
     }
 }
 
 impl std::fmt::Display for ExecutionProofsByRangeRequest {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         write!(
             f,
-            "Request: ExecutionProofsByRange: Start Slot: {}, Count: {}",
-            self.start_slot, self.count
+            "Request: ExecutionProofsByRange: Start Slot: {}, Count: {}, Filters: {}",
+            self.start_slot,
+            self.count,
+            self.proof_filters.len()
         )
     }
 }

beacon_node/lighthouse_network/src/rpc/protocol.rs

@@ -585,10 +585,10 @@ impl ProtocolId {
                 LightClientUpdatesByRangeRequest::ssz_min_len(),
                 LightClientUpdatesByRangeRequest::ssz_max_len(),
             ),
-            Protocol::MetaData => RpcLimits::new(0, 0), // Metadata requests are empty
+            Protocol::MetaData => RpcLimits::new(0, 0),
             Protocol::ExecutionProofsByRange => RpcLimits::new(
                 ExecutionProofsByRangeRequest::ssz_min_len(),
-                ExecutionProofsByRangeRequest::ssz_max_len(),
+                ExecutionProofsByRangeRequest::ssz_max_len(spec.max_request_blocks_upper_bound()),
             ),
             // ExecutionProofsByRoot request is List[ProofByRootIdentifier, MAX_BLOCKS_BY_ROOT.
             Protocol::ExecutionProofsByRoot => RpcLimits::new(0, spec.max_blocks_by_root_request),

beacon_node/network/src/network_beacon_processor/rpc_methods.rs

@@ -1326,7 +1326,12 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
 
     /// Handle an `ExecutionProofsByRange` request from the peer (EIP-8025).
     ///
-    /// Streams all `SignedExecutionProof` objects known for the requested slot range.
+    /// Streams `SignedExecutionProof` objects known for the requested slot range, filtered by
+    /// `proof_filters` when present. For blocks listed in `proof_filters`:
+    /// - a non-empty `proof_types` list → serve only those types
+    /// - an empty `proof_types` list → skip the block entirely (requester already has all proofs)
+    ///
+    /// Blocks absent from `proof_filters` receive all known proof types.
     pub fn handle_execution_proofs_by_range_request(
         &self,
         peer_id: PeerId,
@@ -1351,9 +1356,18 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
             %peer_id,
             start_slot = req.start_slot,
             count = req.count,
+            num_filters = req.proof_filters.len(),
             "Received ExecutionProofsByRange Request"
         );
 
+        // Build a lookup map: block_root → requested proof types from proof_filters.
+        // Blocks not listed in proof_filters will have all known proof types served.
+        let filter_map: std::collections::HashMap<_, _> = req
+            .proof_filters
+            .iter()
+            .map(|id| (id.block_root, &id.proof_types))
+            .collect();
+
         let block_roots = self.get_block_roots_for_slot_range(
             req.start_slot,
             req.count,
@@ -1362,7 +1376,17 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
 
         let mut proofs_sent = 0usize;
         for block_root in block_roots {
+            let allowed_types = filter_map.get(&block_root);
             for proof in self.chain.get_execution_proofs_by_block_root(block_root) {
+                // If this block has a filter entry:
+                //   - empty proof_types  → skip the block entirely (requester already complete)
+                //   - non-empty          → serve only the listed types
+                // An absent entry means "return all types".
+                if let Some(types) = allowed_types
+                    && (types.is_empty() || !types.contains(&proof.message.proof_type))
+                {
+                    continue;
+                }
                 self.send_network_message(NetworkMessage::SendResponse {
                     peer_id,
                     inbound_request_id,

beacon_node/network/src/sync/manager.rs

@@ -353,10 +353,7 @@ impl<T: BeaconChainTypes> SyncManager<T> {
             notified_unknown_roots: LRUTimeCache::new(Duration::from_secs(
                 NOTIFIED_UNKNOWN_ROOT_EXPIRY_SECONDS,
             )),
-            proof_sync: ProofSync::new(
-                beacon_chain.clone(),
-                network_globals.config.proof_sync_activation_slots,
-            ),
+            proof_sync: ProofSync::new(beacon_chain.clone()),
         }
     }
 
@@ -424,14 +421,6 @@ impl<T: BeaconChainTypes> SyncManager<T> {
     #[cfg(test)]
     pub(crate) fn start_proof_sync(&mut self) {
         self.proof_sync.start(&mut self.network);
-        // Advance through the Waiting countdown so callers immediately see Syncing state,
-        // matching pre-Waiting behaviour in unit tests.
-        while matches!(
-            self.proof_sync.state(),
-            super::proof_sync::ProofSyncState::Waiting(_)
-        ) {
-            self.proof_sync.poll(&mut self.network);
-        }
     }
 
     fn network_globals(&self) -> &NetworkGlobals<T::EthSpec> {

beacon_node/network/src/sync/network_context.rs

@@ -48,7 +48,7 @@ use requests::{
 };
 #[cfg(test)]
 use slot_clock::SlotClock;
-use ssz_types::VariableList;
+use ssz_types::{RuntimeVariableList, VariableList};
 use std::collections::hash_map::Entry;
 use std::collections::{HashMap, HashSet};
 use std::fmt::Debug;
@@ -417,17 +417,63 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
 
     /// Send a `ExecutionProofsByRange` request to the given proof-capable peer.
     ///
+    /// `filter_entries` contains `MissingProofInfo` entries for blocks within the requested slot
+    /// range that should appear in `proof_filters`:
+    /// - entries with non-empty `existing_proof_types` → peer returns only the missing types
+    /// - entries with all types already present (`needed` is empty) → peer skips the block
+    ///   entirely (requester already holds all proofs for it)
+    ///
+    /// Blocks with no existing proofs at all are excluded from `filter_entries`; the peer
+    /// will return all known proof types for them by default.
+    ///
     /// Callers should use `find_best_proof_capable_peer` to select the peer first.
     pub fn request_execution_proofs_by_range(
         &mut self,
         peer_id: PeerId,
         start_slot: Slot,
         count: u64,
+        filter_entries: &[MissingProofInfo],
     ) -> Result<ExecutionProofsByRangeRequestId, RpcRequestSendError> {
         let id = ExecutionProofsByRangeRequestId { id: self.next_id() };
+
+        // Build proof_filters from filter_entries:
+        //   - partial blocks: proof_types = types still needed (non-empty)
+        //   - complete blocks: proof_types = [] → peer skips the block entirely
+        //   - fully-missing blocks (existing empty): excluded — peer returns all types by default
+        let max_request_blocks = self
+            .chain
+            .spec
+            .max_request_blocks(self.fork_context.current_fork_name());
+        let mut filter_items: Vec<ProofByRootIdentifier> = Vec::new();
+        for info in filter_entries {
+            if info.existing_proof_types.is_empty() {
+                // Fully missing: no filter entry; peer returns all proof types by default.
+                continue;
+            }
+            let needed: Vec<u8> = self
+                .proof_types
+                .iter()
+                .map(|t| t.to_u8())
+                .filter(|t| !info.existing_proof_types.contains(t))
+                .collect();
+            // needed may be empty for complete blocks — that is intentional.
+            // An empty proof_types list tells the peer to skip this block entirely.
+            let proof_types = VariableList::new(needed)
+                .map_err(|e| RpcRequestSendError::InternalError(format!("proof_types: {e:?}")))?;
+            filter_items.push(ProofByRootIdentifier {
+                block_root: info.root,
+                proof_types,
+            });
+        }
+        let proof_filters =
+            RuntimeVariableList::new(filter_items, max_request_blocks).map_err(|e| {
+                RpcRequestSendError::InternalError(format!("proof_filters too long: {e:?}"))
+            })?;
+
         let request = ExecutionProofsByRangeRequest {
             start_slot: start_slot.as_u64(),
             count,
+            proof_filters,
         };
         self.network_send
             .send(NetworkMessage::SendRequest {
@@ -544,6 +590,14 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         self.network_globals().local_execution_proof_status()
     }
 
+    /// Number of proof types this node is configured to request.
+    ///
+    /// Used by [`ProofSync`] to compute request byte sizes without needing access to the
+    /// full `ProofTypes` set.
+    pub fn configured_proof_types_count(&self) -> usize {
+        self.proof_types.len()
+    }
+
     /// Returns `true` if the peer has `execution_proof_enabled()` in their ENR.
     pub fn is_proof_capable_peer(&self, peer_id: &PeerId) -> bool {
         self.network_globals()