Sync fixes for fusaka-devnet-4

beacon_node/http_api/src/lib.rs

@@ -147,7 +147,6 @@ pub struct Config {
     pub enable_beacon_processor: bool,
     #[serde(with = "eth2::types::serde_status_code")]
     pub duplicate_block_status_code: StatusCode,
-    pub target_peers: usize,
 }
 
 impl Default for Config {
@@ -162,7 +161,6 @@ impl Default for Config {
             sse_capacity_multiplier: 1,
             enable_beacon_processor: true,
             duplicate_block_status_code: StatusCode::ACCEPTED,
-            target_peers: 100,
         }
     }
 }

beacon_node/lighthouse_network/src/config.rs

@@ -341,7 +341,7 @@ impl Default for Config {
             enr_udp6_port: None,
             enr_quic6_port: None,
             enr_tcp6_port: None,
-            target_peers: 100,
+            target_peers: 200,
             discv5_config,
             boot_nodes_enr: vec![],
             boot_nodes_multiaddr: vec![],

beacon_node/lighthouse_network/src/peer_manager/peerdb.rs

@@ -300,6 +300,7 @@ impl<E: EthSpec> PeerDB<E> {
             .filter(move |(_, info)| {
                 // We check both the metadata and gossipsub data as we only want to count long-lived subscribed peers
                 info.is_connected()
+                    && info.is_synced_or_advanced()
                     && info.on_subnet_metadata(&subnet)
                     && info.on_subnet_gossipsub(&subnet)
                     && info.is_good_gossipsub_peer()
@@ -318,7 +319,43 @@ impl<E: EthSpec> PeerDB<E> {
             .filter(move |(_, info)| {
                 // The custody_subnets hashset can be populated via enr or metadata
                 let is_custody_subnet_peer = info.is_assigned_to_custody_subnet(&subnet);
-                info.is_connected() && info.is_good_gossipsub_peer() && is_custody_subnet_peer
+                info.is_connected()
+                    && info.is_good_gossipsub_peer()
+                    && is_custody_subnet_peer
+                    && matches!(
+                        info.sync_status(),
+                        SyncStatus::Synced { .. } | SyncStatus::Advanced { .. }
+                    )
+            })
+            .map(|(peer_id, _)| peer_id)
+    }
+
+    /// Returns an iterator of all good gossipsub peers that are supposed to be custodying
+    /// the given subnet id.
+    pub fn good_custody_subnet_peer_range_sync(
+        &self,
+        subnet: DataColumnSubnetId,
+        epoch: Epoch,
+    ) -> impl Iterator<Item = &PeerId> {
+        self.peers
+            .iter()
+            .filter(move |(_, info)| {
+                // The custody_subnets hashset can be populated via enr or metadata
+                let is_custody_subnet_peer = info.is_assigned_to_custody_subnet(&subnet);
+
+                info.is_connected()
+                    && is_custody_subnet_peer
+                    && match info.sync_status() {
+                        SyncStatus::Synced { info } => {
+                            info.has_slot(epoch.end_slot(E::slots_per_epoch()))
+                        }
+                        SyncStatus::Advanced { info } => {
+                            info.has_slot(epoch.end_slot(E::slots_per_epoch()))
+                        }
+                        SyncStatus::IrrelevantPeer
+                        | SyncStatus::Behind { .. }
+                        | SyncStatus::Unknown => false,
+                    }
             })
             .map(|(peer_id, _)| peer_id)
     }
@@ -333,7 +370,9 @@ impl<E: EthSpec> PeerDB<E> {
             .iter()
             .filter(move |(_, info)| {
                 // The custody_subnets hashset can be populated via enr or metadata
-                info.is_connected() && info.is_assigned_to_custody_subnet(&subnet)
+                info.is_connected()
+                    && info.is_synced_or_advanced()
+                    && info.is_assigned_to_custody_subnet(&subnet)
             })
             .map(|(peer_id, _)| peer_id)
     }

beacon_node/lighthouse_network/src/peer_manager/peerdb/peer_info.rs

@@ -176,7 +176,7 @@ impl<E: EthSpec> PeerInfo<E> {
 
     /// Returns the number of long lived subnets a peer is subscribed to.
     // NOTE: This currently excludes sync committee subnets
-    pub fn long_lived_subnet_count(&self) -> usize {
+    pub fn long_lived_attnet_count(&self) -> usize {
         if let Some(meta_data) = self.meta_data.as_ref() {
             return meta_data.attnets().num_set_bits();
         } else if let Some(enr) = self.enr.as_ref()
@@ -222,6 +222,13 @@ impl<E: EthSpec> PeerInfo<E> {
                 }
             }
         }
+
+        long_lived_subnets.extend(
+            self.custody_subnets
+                .iter()
+                .map(|&id| Subnet::DataColumn(id)),
+        );
+
         long_lived_subnets
     }
 
@@ -262,6 +269,11 @@ impl<E: EthSpec> PeerInfo<E> {
         {
             return true;
         }
+
+        if !self.custody_subnets.is_empty() {
+            return true;
+        }
+
         false
     }
 
@@ -318,6 +330,14 @@ impl<E: EthSpec> PeerInfo<E> {
         )
     }
 
+    /// Checks if the peer is synced or advanced.
+    pub fn is_synced_or_advanced(&self) -> bool {
+        matches!(
+            self.sync_status,
+            SyncStatus::Synced { .. } | SyncStatus::Advanced { .. }
+        )
+    }
+
     /// Checks if the status is connected.
     pub fn is_dialing(&self) -> bool {
         matches!(self.connection_status, PeerConnectionStatus::Dialing { .. })

beacon_node/lighthouse_network/src/rpc/protocol.rs

@@ -825,8 +825,8 @@ impl<E: EthSpec> RequestType<E> {
         match self {
             // add more protocols when versions/encodings are supported
             RequestType::Status(_) => vec![
-                ProtocolId::new(SupportedProtocol::StatusV1, Encoding::SSZSnappy),
                 ProtocolId::new(SupportedProtocol::StatusV2, Encoding::SSZSnappy),
+                ProtocolId::new(SupportedProtocol::StatusV1, Encoding::SSZSnappy),
             ],
             RequestType::Goodbye(_) => vec![ProtocolId::new(
                 SupportedProtocol::GoodbyeV1,

beacon_node/lighthouse_network/src/service/mod.rs

@@ -1911,7 +1911,7 @@ impl<E: EthSpec> Network<E> {
                         }
                     },
                 };
-                debug!(our_addr = %local_addr, from = %send_back_addr, error = error_repr, "Failed incoming connection");
+                tracing::trace!(our_addr = %local_addr, from = %send_back_addr, error = error_repr, "Failed incoming connection");
                 None
             }
             SwarmEvent::OutgoingConnectionError {

beacon_node/lighthouse_network/tests/rpc_tests.rs

@@ -76,21 +76,23 @@ fn test_tcp_status_rpc() {
         .await;
 
         // Dummy STATUS RPC message
-        let rpc_request = RequestType::Status(StatusMessage::V1(StatusMessageV1 {
+        let rpc_request = RequestType::Status(StatusMessage::V2(StatusMessageV2 {
             fork_digest: [0; 4],
             finalized_root: Hash256::zero(),
             finalized_epoch: Epoch::new(1),
             head_root: Hash256::zero(),
             head_slot: Slot::new(1),
+            earliest_available_slot: Slot::new(0),
         }));
 
         // Dummy STATUS RPC message
-        let rpc_response = Response::Status(StatusMessage::V1(StatusMessageV1 {
+        let rpc_response = Response::Status(StatusMessage::V2(StatusMessageV2 {
             fork_digest: [0; 4],
             finalized_root: Hash256::zero(),
             finalized_epoch: Epoch::new(1),
             head_root: Hash256::zero(),
             head_slot: Slot::new(1),
+            earliest_available_slot: Slot::new(0),
         }));
 
         // build the sender future
@@ -1205,21 +1207,23 @@ fn test_delayed_rpc_response() {
         .await;
 
         // Dummy STATUS RPC message
-        let rpc_request = RequestType::Status(StatusMessage::V1(StatusMessageV1 {
+        let rpc_request = RequestType::Status(StatusMessage::V2(StatusMessageV2 {
             fork_digest: [0; 4],
             finalized_root: Hash256::from_low_u64_be(0),
             finalized_epoch: Epoch::new(1),
             head_root: Hash256::from_low_u64_be(0),
             head_slot: Slot::new(1),
+            earliest_available_slot: Slot::new(0),
         }));
 
         // Dummy STATUS RPC message
-        let rpc_response = Response::Status(StatusMessage::V1(StatusMessageV1 {
+        let rpc_response = Response::Status(StatusMessage::V2(StatusMessageV2 {
             fork_digest: [0; 4],
             finalized_root: Hash256::from_low_u64_be(0),
             finalized_epoch: Epoch::new(1),
             head_root: Hash256::from_low_u64_be(0),
             head_slot: Slot::new(1),
+            earliest_available_slot: Slot::new(0),
         }));
 
         // build the sender future
@@ -1335,21 +1339,23 @@ fn test_active_requests() {
         .await;
 
         // Dummy STATUS RPC request.
-        let rpc_request = RequestType::Status(StatusMessage::V1(StatusMessageV1 {
+        let rpc_request = RequestType::Status(StatusMessage::V2(StatusMessageV2 {
             fork_digest: [0; 4],
             finalized_root: Hash256::from_low_u64_be(0),
             finalized_epoch: Epoch::new(1),
             head_root: Hash256::from_low_u64_be(0),
             head_slot: Slot::new(1),
+            earliest_available_slot: Slot::new(0),
         }));
 
         // Dummy STATUS RPC response.
-        let rpc_response = Response::Status(StatusMessage::V1(StatusMessageV1 {
+        let rpc_response = Response::Status(StatusMessage::V2(StatusMessageV2 {
             fork_digest: [0; 4],
             finalized_root: Hash256::zero(),
             finalized_epoch: Epoch::new(1),
             head_root: Hash256::zero(),
             head_slot: Slot::new(1),
+            earliest_available_slot: Slot::new(0),
         }));
 
         // Number of requests.

beacon_node/network/src/network_beacon_processor/rpc_methods.rs

@@ -212,7 +212,7 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                     send_block_count += 1;
                 }
                 Ok(None) => {
-                    debug!(
+                    tracing::trace!(
                         %peer_id,
                         request_root = ?root,
                         "Peer requested unknown block"

beacon_node/network/src/network_beacon_processor/sync_methods.rs

@@ -10,7 +10,8 @@ use beacon_chain::data_availability_checker::AvailabilityCheckError;
 use beacon_chain::data_availability_checker::MaybeAvailableBlock;
 use beacon_chain::{
     AvailabilityProcessingStatus, BeaconChainTypes, BlockError, ChainSegmentResult,
-    HistoricalBlockError, NotifyExecutionLayer, validator_monitor::get_slot_delay_ms,
+    ExecutionPayloadError, HistoricalBlockError, NotifyExecutionLayer,
+    validator_monitor::get_slot_delay_ms,
 };
 use beacon_processor::{
     AsyncFn, BlockingFn, DuplicateCache,
@@ -774,7 +775,19 @@ impl<T: BeaconChainTypes> NetworkBeaconProcessor<T> {
                 })
             }
             ref err @ BlockError::ExecutionPayloadError(ref epe) => {
-                if !epe.penalize_peer() {
+                if matches!(epe, ExecutionPayloadError::RejectedByExecutionEngine { .. }) {
+                    debug!(
+                        error = ?err,
+                        "Invalid execution payload rejected by EE"
+                    );
+                    Err(ChainSegmentFailed {
+                        message: format!(
+                            "Peer sent a block containing invalid execution payload. Reason: {:?}",
+                            err
+                        ),
+                        peer_action: Some(PeerAction::LowToleranceError),
+                    })
+                } else if !epe.penalize_peer() {
                     // These errors indicate an issue with the EL and not the `ChainSegment`.
                     // Pause the syncing while the EL recovers
                     debug!(

beacon_node/network/src/sync/backfill_sync/mod.rs

@@ -313,7 +313,6 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
                     CouplingError::DataColumnPeerFailure {
                         error,
                         faulty_peers,
-                        action,
                         exceeded_retries,
                     } => {
                         debug!(?batch_id, error, "Block components coupling error");
@@ -325,9 +324,6 @@ impl<T: BeaconChainTypes> BackFillSync<T> {
                             failed_columns.insert(*column);
                             failed_peers.insert(*peer);
                         }
-                        for peer in failed_peers.iter() {
-                            network.report_peer(*peer, *action, "failed to return columns");
-                        }
 
                         // Only retry if peer failure **and** retries have been exceeded
                         if !*exceeded_retries {

beacon_node/network/src/sync/block_sidecar_coupling.rs

@@ -2,7 +2,7 @@ use beacon_chain::{
     block_verification_types::RpcBlock, data_column_verification::CustodyDataColumn, get_block_root,
 };
 use lighthouse_network::{
-    PeerAction, PeerId,
+    PeerId,
     service::api_types::{
         BlobsByRangeRequestId, BlocksByRangeRequestId, DataColumnsByRangeRequestId,
     },
@@ -60,7 +60,6 @@ pub(crate) enum CouplingError {
     DataColumnPeerFailure {
         error: String,
         faulty_peers: Vec<(ColumnIndex, PeerId)>,
-        action: PeerAction,
         exceeded_retries: bool,
     },
     BlobPeerFailure(String),
@@ -249,7 +248,6 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                 if let Err(CouplingError::DataColumnPeerFailure {
                     error: _,
                     faulty_peers,
-                    action: _,
                     exceeded_retries: _,
                 }) = &resp
                 {
@@ -374,7 +372,6 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                     return Err(CouplingError::DataColumnPeerFailure {
                         error: format!("No columns for block {block_root:?} with data"),
                         faulty_peers: responsible_peers,
-                        action: PeerAction::LowToleranceError,
                         exceeded_retries,
 
                     });
@@ -399,7 +396,6 @@ impl<E: EthSpec> RangeBlockComponentsRequest<E> {
                     return Err(CouplingError::DataColumnPeerFailure {
                         error: format!("Peers did not return column for block_root {block_root:?} {naughty_peers:?}"),
                         faulty_peers: naughty_peers,
-                        action: PeerAction::LowToleranceError,
                         exceeded_retries
                     });
                 }
@@ -465,7 +461,7 @@ mod tests {
         NumBlobs, generate_rand_block_and_blobs, generate_rand_block_and_data_columns, test_spec,
     };
     use lighthouse_network::{
-        PeerAction, PeerId,
+        PeerId,
         service::api_types::{
             BlobsByRangeRequestId, BlocksByRangeRequestId, ComponentsByRangeRequestId,
             DataColumnsByRangeRequestId, Id, RangeRequestId,
@@ -773,15 +769,13 @@ mod tests {
         if let Err(super::CouplingError::DataColumnPeerFailure {
             error,
             faulty_peers,
-            action,
             exceeded_retries,
         }) = result
         {
             assert!(error.contains("Peers did not return column"));
             assert_eq!(faulty_peers.len(), 2); // columns 3 and 4 missing
             assert_eq!(faulty_peers[0].0, 3); // column index 3
             assert_eq!(faulty_peers[1].0, 4); // column index 4
-            assert!(matches!(action, PeerAction::LowToleranceError));
             assert!(!exceeded_retries); // First attempt, should be false
         } else {
             panic!("Expected PeerFailure error");
@@ -943,13 +937,11 @@ mod tests {
         if let Err(super::CouplingError::DataColumnPeerFailure {
             error: _,
             faulty_peers,
-            action,
             exceeded_retries,
         }) = result
         {
             assert_eq!(faulty_peers.len(), 1); // column 2 missing
             assert_eq!(faulty_peers[0].0, 2); // column index 2
-            assert!(matches!(action, PeerAction::LowToleranceError));
             assert!(exceeded_retries); // Should be true after max retries
         } else {
             panic!("Expected PeerFailure error with exceeded_retries=true");

beacon_node/network/src/sync/network_context.rs

@@ -715,7 +715,6 @@ impl<T: BeaconChainTypes> SyncNetworkContext<T> {
         let range_req = entry.get_mut();
         if let Some(blocks_result) = range_req.responses(&self.chain.spec) {
             if let Err(CouplingError::DataColumnPeerFailure {
-                action: _,
                 error,
                 faulty_peers: _,
                 exceeded_retries,