peer reputation with separate app & gossipsub score (#1028)

The approach is mostly detailed in
[Notion](https://www.notion.so/fuellabs/New-Proposal-96da9de6609b4faf94c30ae6bd355bc4).

These are the first steps of our Peer Reputation approach, there are
certain _todos_ left, could be added later on or if needed I can add
them with this PR.

Todo:
- [x] peer reputation decay with time
- [ ] implement unban peers after certain period
- [x] reserved peers should not be banned/scored ? - this is kind of
"implemented" by only updating the score of non-reserved peers


Ref FuelLabs/fuel-core#943

---------

Co-authored-by: ControlCplusControlV <44706811+ControlCplusControlV@users.noreply.github.com>
Co-authored-by: green <xgreenx9999@gmail.com>
Co-authored-by: Brandon Kite <brandonkite92@gmail.com>

crates/services/p2p/src/behavior.rs

@@ -1,8 +1,3 @@
-use std::sync::{
-    Arc,
-    RwLock,
-};
-
 use crate::{
     codecs::NetworkCodec,
     config::Config,
@@ -15,11 +10,9 @@ use crate::{
         config::build_gossipsub_behaviour,
         topics::GossipTopic,
     },
-    peer_manager::{
-        ConnectionState,
-        PeerInfo,
-        PeerInfoEvent,
-        PeerManagerBehaviour,
+    peer_report::{
+        PeerReportBehaviour,
+        PeerReportEvent,
     },
     request_response::messages::{
         NetworkResponse,
@@ -29,10 +22,7 @@ use crate::{
 use fuel_core_types::fuel_types::BlockHeight;
 use libp2p::{
     gossipsub::{
-        error::{
-            PublishError,
-            SubscriptionError,
-        },
+        error::PublishError,
         Gossipsub,
         GossipsubEvent,
         MessageAcceptance,
@@ -50,11 +40,16 @@ use libp2p::{
     Multiaddr,
     PeerId,
 };
+use tracing::{
+    debug,
+    error,
+    log::warn,
+};
 
 #[derive(Debug)]
 pub enum FuelBehaviourEvent {
     Discovery(DiscoveryEvent),
-    PeerInfo(PeerInfoEvent),
+    PeerReport(PeerReportEvent),
     Gossipsub(GossipsubEvent),
     RequestResponse(RequestResponseEvent<RequestMessage, NetworkResponse>),
 }
@@ -66,9 +61,8 @@ pub struct FuelBehaviour<Codec: NetworkCodec> {
     /// Node discovery
     discovery: DiscoveryBehaviour,
 
-    /// Handles Peer Connections
     /// Identifies and periodically requests `BlockHeight` from connected nodes
-    peer_manager: PeerManagerBehaviour,
+    peer_report: PeerReportBehaviour,
 
     /// Message propagation for p2p
     gossipsub: Gossipsub,
@@ -78,11 +72,7 @@ pub struct FuelBehaviour<Codec: NetworkCodec> {
 }
 
 impl<Codec: NetworkCodec> FuelBehaviour<Codec> {
-    pub(crate) fn new(
-        p2p_config: &Config,
-        codec: Codec,
-        connection_state: Arc<RwLock<ConnectionState>>,
-    ) -> Self {
+    pub(crate) fn new(p2p_config: &Config, codec: Codec) -> Self {
         let local_public_key = p2p_config.keypair.public();
         let local_peer_id = PeerId::from_public_key(&local_public_key);
 
@@ -109,7 +99,9 @@ impl<Codec: NetworkCodec> FuelBehaviour<Codec> {
             discovery_config
         };
 
-        let peer_manager = PeerManagerBehaviour::new(p2p_config, connection_state);
+        let gossipsub = build_gossipsub_behaviour(p2p_config);
+
+        let peer_report = PeerReportBehaviour::new(p2p_config);
 
         let req_res_protocol =
             std::iter::once((codec.get_req_res_protocol(), ProtocolSupport::Full));
@@ -123,20 +115,12 @@ impl<Codec: NetworkCodec> FuelBehaviour<Codec> {
 
         Self {
             discovery: discovery_config.finish(),
-            gossipsub: build_gossipsub_behaviour(p2p_config),
-            peer_manager,
+            gossipsub,
+            peer_report,
             request_response,
         }
     }
 
-    pub fn add_addresses_to_peer_info(
-        &mut self,
-        peer_id: &PeerId,
-        addresses: Vec<Multiaddr>,
-    ) {
-        self.peer_manager.insert_peer_addresses(peer_id, addresses);
-    }
-
     pub fn add_addresses_to_discovery(
         &mut self,
         peer_id: &PeerId,
@@ -147,14 +131,6 @@ impl<Codec: NetworkCodec> FuelBehaviour<Codec> {
         }
     }
 
-    pub fn get_peers_ids(&self) -> impl Iterator<Item = &PeerId> {
-        self.peer_manager.get_peers_ids()
-    }
-
-    pub fn total_peers_connected(&self) -> usize {
-        self.peer_manager.total_peers_connected()
-    }
-
     pub fn publish_message(
         &mut self,
         topic: GossipTopic,
@@ -163,13 +139,6 @@ impl<Codec: NetworkCodec> FuelBehaviour<Codec> {
         self.gossipsub.publish(topic, encoded_data)
     }
 
-    pub fn subscribe_to_topic(
-        &mut self,
-        topic: &GossipTopic,
-    ) -> Result<bool, SubscriptionError> {
-        self.gossipsub.subscribe(topic)
-    }
-
     pub fn send_request_msg(
         &mut self,
         message_request: RequestMessage,
@@ -191,26 +160,33 @@ impl<Codec: NetworkCodec> FuelBehaviour<Codec> {
         msg_id: &MessageId,
         propagation_source: &PeerId,
         acceptance: MessageAcceptance,
-    ) -> Result<bool, PublishError> {
-        self.gossipsub.report_message_validation_result(
+    ) -> Option<f64> {
+        let should_check_score = matches!(acceptance, MessageAcceptance::Reject);
+
+        match self.gossipsub.report_message_validation_result(
             msg_id,
             propagation_source,
             acceptance,
-        )
-    }
-
-    pub fn update_block_height(&mut self, block_height: BlockHeight) {
-        self.peer_manager.update_block_height(block_height);
-    }
+        ) {
+            Ok(true) => {
+                debug!(target: "fuel-p2p", "Sent a report for MessageId: {} from PeerId: {}", msg_id, propagation_source);
+                if should_check_score {
+                    return self.gossipsub.peer_score(propagation_source)
+                }
+            }
+            Ok(false) => {
+                warn!(target: "fuel-p2p", "Message with MessageId: {} not found in the Gossipsub Message Cache", msg_id);
+            }
+            Err(e) => {
+                error!(target: "fuel-p2p", "Failed to report Message with MessageId: {} with Error: {:?}", msg_id, e);
+            }
+        }
 
-    // Currently only used in testing, but should be useful for the P2P Service API
-    #[allow(dead_code)]
-    pub fn get_peer_info(&self, peer_id: &PeerId) -> Option<&PeerInfo> {
-        self.peer_manager.get_peer_info(peer_id)
+        None
     }
 
-    pub fn peer_manager(&self) -> &PeerManagerBehaviour {
-        &self.peer_manager
+    pub fn update_block_height(&mut self, block_height: BlockHeight) {
+        self.peer_report.update_block_height(block_height);
     }
 }
 
@@ -220,9 +196,9 @@ impl From<DiscoveryEvent> for FuelBehaviourEvent {
     }
 }
 
-impl From<PeerInfoEvent> for FuelBehaviourEvent {
-    fn from(event: PeerInfoEvent) -> Self {
-        FuelBehaviourEvent::PeerInfo(event)
+impl From<PeerReportEvent> for FuelBehaviourEvent {
+    fn from(event: PeerReportEvent) -> Self {
+        FuelBehaviourEvent::PeerReport(event)
     }
 }