feat(node): notify peer it is now considered as BAD

sn_networking/src/cmd.rs

@@ -735,6 +735,9 @@ impl SwarmDriver {
         info!("Peer {peer_id:?} is reported as having issue {issue:?}");
         let (issue_vec, is_bad) = self.bad_nodes.entry(peer_id).or_default();
 
+        let mut is_new_bad = false;
+        let mut bad_behaviour: String = "".to_string();
+
         // If being considered as bad already, skip certain operations
         if !(*is_bad) {
             // Remove outdated entries
@@ -767,6 +770,8 @@ impl SwarmDriver {
                     .count();
                 if issue_counts >= 3 {
                     *is_bad = true;
+                    is_new_bad = true;
+                    bad_behaviour = format!("{issue:?}");
                     info!("Peer {peer_id:?} accumulated {issue_counts} times of issue {issue:?}. Consider it as a bad node now.");
                     // Once a bad behaviour detected, no point to continue
                     break;
@@ -785,6 +790,14 @@ impl SwarmDriver {
                 self.log_kbuckets(&peer_id);
                 let _ = self.check_for_change_in_our_close_group();
             }
+
+            if is_new_bad {
+                self.send_event(NetworkEvent::PeerConsideredAsBad {
+                    detected_by: self.self_peer_id,
+                    bad_peer: peer_id,
+                    bad_behaviour,
+                });
+            }
         }
     }
 

sn_networking/src/event.rs

@@ -114,6 +114,12 @@ pub enum NetworkEvent {
         our_protocol: String,
         their_protocol: String,
     },
+    /// The peer is now considered as a bad node, due to the detected bad behaviour
+    PeerConsideredAsBad {
+        detected_by: PeerId,
+        bad_peer: PeerId,
+        bad_behaviour: String,
+    },
     /// The records bearing these keys are to be fetched from the holder or the network
     KeysToFetchForReplication(Vec<(PeerId, RecordKey)>),
     /// Started listening on a new address
@@ -160,6 +166,16 @@ impl Debug for NetworkEvent {
             } => {
                 write!(f, "NetworkEvent::PeerWithUnsupportedProtocol({our_protocol:?}, {their_protocol:?})")
             }
+            NetworkEvent::PeerConsideredAsBad {
+                bad_peer,
+                bad_behaviour,
+                ..
+            } => {
+                write!(
+                    f,
+                    "NetworkEvent::PeerConsideredAsBad({bad_peer:?}, {bad_behaviour:?})"
+                )
+            }
             NetworkEvent::KeysToFetchForReplication(list) => {
                 let keys_len = list.len();
                 write!(f, "NetworkEvent::KeysForReplication({keys_len:?})")
@@ -695,6 +711,28 @@ impl SwarmDriver {
                                 .collect();
                             self.send_event(NetworkEvent::QuoteVerification { quotes })
                         }
+                        Request::Cmd(sn_protocol::messages::Cmd::PeerConsideredAsBad {
+                            detected_by,
+                            bad_peer,
+                            bad_behaviour,
+                        }) => {
+                            let response = Response::Cmd(
+                                sn_protocol::messages::CmdResponse::PeerConsideredAsBad(Ok(())),
+                            );
+                            self.swarm
+                                .behaviour_mut()
+                                .request_response
+                                .send_response(channel, response)
+                                .map_err(|_| NetworkError::InternalMsgChannelDropped)?;
+
+                            if bad_peer == NetworkAddress::from_peer(self.self_peer_id) {
+                                warn!("Peer {detected_by:?} consider us as BAD, due to {bad_behaviour:?}.");
+                                // TODO: shall we terminate self after received such notifications
+                                //       from the majority close_group nodes around us?
+                            } else {
+                                error!("Received a bad_peer notification from {detected_by:?}, targeting {bad_peer:?}, which is not us.");
+                            }
+                        }
                         Request::Query(query) => {
                             self.send_event(NetworkEvent::QueryRequestReceived {
                                 query,

sn_node/src/node.rs

@@ -27,7 +27,7 @@ use sn_networking::{
 };
 use sn_protocol::{
     error::Error as ProtocolError,
-    messages::{ChunkProof, CmdResponse, Query, QueryResponse, Request, Response},
+    messages::{ChunkProof, Cmd, CmdResponse, Query, QueryResponse, Request, Response},
     NetworkAddress, PrettyPrintRecordKey,
 };
 use sn_transfers::{HotWallet, MainPubkey, MainSecretKey, NanoTokens};
@@ -329,6 +329,23 @@ impl Node {
             NetworkEvent::PeerWithUnsupportedProtocol { .. } => {
                 event_header = "PeerWithUnsupportedProtocol";
             }
+            NetworkEvent::PeerConsideredAsBad {
+                detected_by,
+                bad_peer,
+                bad_behaviour,
+            } => {
+                event_header = "PeerConsideredAsBad";
+                let request = Request::Cmd(Cmd::PeerConsideredAsBad {
+                    detected_by: NetworkAddress::from_peer(detected_by),
+                    bad_peer: NetworkAddress::from_peer(bad_peer),
+                    bad_behaviour,
+                });
+
+                let network = self.network.clone();
+                let _handle = spawn(async move {
+                    network.send_req_ignore_reply(request, bad_peer);
+                });
+            }
             NetworkEvent::NewListenAddr(_) => {
                 event_header = "NewListenAddr";
                 if !cfg!(feature = "local-discovery") {

sn_protocol/src/messages/cmd.rs

@@ -35,6 +35,12 @@ pub enum Cmd {
         target: NetworkAddress,
         quotes: Vec<(NetworkAddress, PaymentQuote)>,
     },
+    /// Notify the peer it is now being considered as BAD due to the included behaviour
+    PeerConsideredAsBad {
+        detected_by: NetworkAddress,
+        bad_peer: NetworkAddress,
+        bad_behaviour: String,
+    },
 }
 
 impl std::fmt::Debug for Cmd {
@@ -53,6 +59,16 @@ impl std::fmt::Debug for Cmd {
                 .field("target", target)
                 .field("quotes_len", &quotes.len())
                 .finish(),
+            Cmd::PeerConsideredAsBad {
+                detected_by,
+                bad_peer,
+                bad_behaviour,
+            } => f
+                .debug_struct("Cmd::PeerConsideredAsBad")
+                .field("detected_by", detected_by)
+                .field("bad_peer", bad_peer)
+                .field("bad_behaviour", bad_behaviour)
+                .finish(),
         }
     }
 }
@@ -63,6 +79,7 @@ impl Cmd {
         match self {
             Cmd::Replicate { holder, .. } => holder.clone(),
             Cmd::QuoteVerification { target, .. } => target.clone(),
+            Cmd::PeerConsideredAsBad { bad_peer, .. } => bad_peer.clone(),
         }
     }
 }
@@ -85,6 +102,15 @@ impl std::fmt::Display for Cmd {
                     quotes.len()
                 )
             }
+            Cmd::PeerConsideredAsBad {
+                detected_by,
+                bad_peer,
+                bad_behaviour,
+            } => {
+                write!(
+                    f,
+                    "Cmd::PeerConsideredAsBad({detected_by:?} consider peer {bad_peer:?} as bad, due to {bad_behaviour:?})")
+            }
         }
     }
 }

sn_protocol/src/messages/response.rs

@@ -111,6 +111,11 @@ pub enum CmdResponse {
     //
     /// Response to quote verification cmd
     QuoteVerification(Result<()>),
+    //
+    // ===== PeerConsideredAsBad =====
+    //
+    /// Response to the considered as bad notification
+    PeerConsideredAsBad(Result<()>),
 }
 
 /// The Ok variant of a CmdResponse