Downscore timed-out peers when syncing

grandinetech · Jul 15, 2024 · 1148eee · 1148eee
1 parent be97b0e
commit 1148eee
Show file tree

Hide file tree

Showing 6 changed files with 32 additions and 6 deletions.
diff --git a/Cargo.lock b/Cargo.lock
diff --git a/p2p/Cargo.toml b/p2p/Cargo.toml
@@ -36,6 +36,7 @@ slog = { workspace = true }
 slog-stdlog = { workspace = true }
 ssz = { workspace = true }
 std_ext = { workspace = true }
+strum = { workspace = true }
 thiserror = { workspace = true }
 tokio = { workspace = true }
 tokio-stream = { workspace = true }

diff --git a/p2p/src/block_sync_service.rs b/p2p/src/block_sync_service.rs
@@ -4,7 +4,7 @@ use std::sync::Arc;
 use anyhow::Result;
 use database::Database;
 use eth1_api::RealController;
-use eth2_libp2p::{rpc::StatusMessage, PeerId};
+use eth2_libp2p::{rpc::StatusMessage, PeerAction, PeerId, ReportSource};
 use fork_choice_control::SyncMessage;
 use futures::{
     channel::mpsc::{UnboundedReceiver, UnboundedSender},
@@ -29,7 +29,7 @@ use types::{
 use crate::{
     back_sync::{BackSync, Data as BackSyncData, Error as BackSyncError, SyncCheckpoint},
     messages::{ArchiverToSync, P2pToSync, SyncToApi, SyncToMetrics, SyncToP2p},
-    misc::RequestId,
+    misc::{PeerReportReason, RequestId},
     sync_manager::{SyncBatch, SyncManager, SyncTarget},
 };
 
@@ -354,11 +354,20 @@ impl<P: Preset> BlockSyncService<P> {
             let request_id = self.request_id()?;
             let SyncBatch {
                 target,
+                peer_id,
                 start_slot,
                 count,
                 ..
             } = batch;
 
+            SyncToP2p::ReportPeer(
+                peer_id,
+                PeerAction::MidToleranceError,
+                ReportSource::SyncService,
+                PeerReportReason::ExpiredSyncBatch,
+            )
+            .send(&self.sync_to_p2p_tx);
+
             let peer = self.sync_manager.retry_batch(request_id, &batch);
 
             if let Some(peer_id) = peer {

diff --git a/p2p/src/messages.rs b/p2p/src/messages.rs
@@ -26,7 +26,7 @@ use types::{
 
 use crate::{
     misc::{
-        AttestationSubnetActions, BeaconCommitteeSubscription, RequestId,
+        AttestationSubnetActions, BeaconCommitteeSubscription, PeerReportReason, RequestId,
         SyncCommitteeSubnetAction, SyncCommitteeSubscription,
     },
     network_api::{NodeIdentity, NodePeer, NodePeerCount, NodePeersQuery},
@@ -108,6 +108,7 @@ impl SyncToMetrics {
 
 pub enum SyncToP2p {
     PruneReceivedBlocks,
+    ReportPeer(PeerId, PeerAction, ReportSource, PeerReportReason),
     RequestBlobsByRange(RequestId, PeerId, Slot, u64),
     RequestBlobsByRoot(RequestId, PeerId, Vec<BlobIdentifier>),
     RequestBlocksByRange(RequestId, PeerId, Slot, u64),

diff --git a/p2p/src/misc.rs b/p2p/src/misc.rs
@@ -2,6 +2,7 @@ use std::collections::BTreeMap;
 
 use serde::{Deserialize, Serialize};
 use serde_with::{As, DisplayFromStr};
+use strum::IntoStaticStr;
 use types::phase0::primitives::{CommitteeIndex, Epoch, Slot, SubnetId, ValidatorIndex};
 
 pub type RequestId = usize;
@@ -52,6 +53,11 @@ pub struct BeaconCommitteeSubscription {
     pub is_aggregator: bool,
 }
 
+#[derive(IntoStaticStr)]
+pub enum PeerReportReason {
+    ExpiredSyncBatch,
+}
+
 #[derive(PartialEq, Eq, Debug, Deserialize, Serialize)]
 #[serde(deny_unknown_fields)]
 pub struct SyncCommitteeSubscription {

diff --git a/p2p/src/network.rs b/p2p/src/network.rs
@@ -390,6 +390,17 @@ impl<P: Preset> Network<P> {
 
                 message = self.channels.sync_to_p2p_rx.select_next_some() => {
                     match message {
+                        SyncToP2p::PruneReceivedBlocks => {
+                            self.received_block_roots = HashMap::new();
+                        }
+                        SyncToP2p::ReportPeer(peer_id, peer_action, report_source, reason) => {
+                            self.report_peer(
+                                peer_id,
+                                peer_action,
+                                report_source,
+                                reason
+                            );
+                        }
                         SyncToP2p::RequestBlobsByRange(request_id, peer_id, start_slot, count) => {
                             self.request_blobs_by_range(request_id, peer_id, start_slot, count);
                         }
@@ -408,9 +419,6 @@ impl<P: Preset> Network<P> {
                         SyncToP2p::SubscribeToCoreTopics => {
                             self.subscribe_to_core_topics();
                         }
-                        SyncToP2p::PruneReceivedBlocks => {
-                            self.received_block_roots = HashMap::new();
-                        }
                     }
                 },