Introduce sc_peerset::DropReason (#7996)

* Introduce sc_peerset::DropReason

* Fix peerset tests

Author: tomaka

client/network/src/protocol/generic_proto/behaviour.rs

@@ -469,7 +469,7 @@ impl GenericProto {
 				timer: _
 			} => {
 				debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
-				self.peerset.dropped(set_id, peer_id.clone());
+				self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
 				let backoff_until = Some(if let Some(ban) = ban {
 					cmp::max(timer_deadline, Instant::now() + ban)
 				} else {
@@ -486,7 +486,7 @@ impl GenericProto {
 			// If relevant, the external API is instantly notified.
 			PeerState::Enabled { mut connections } => {
 				debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
-				self.peerset.dropped(set_id, peer_id.clone());
+				self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
 
 				if connections.iter().any(|(_, s)| matches!(s, ConnectionState::Open(_))) {
 					debug!(target: "sub-libp2p", "External API <= Closed({}, {:?})", peer_id, set_id);
@@ -942,7 +942,7 @@ impl GenericProto {
 				_ => {
 					debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})",
 						incoming.peer_id, incoming.set_id);
-					self.peerset.dropped(incoming.set_id, incoming.peer_id);
+					self.peerset.dropped(incoming.set_id, incoming.peer_id, sc_peerset::DropReason::Unknown);
 				},
 			}
 			return
@@ -1184,7 +1184,7 @@ impl NetworkBehaviour for GenericProto {
 
 					if connections.is_empty() {
 						debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
-						self.peerset.dropped(set_id, peer_id.clone());
+						self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
 						*entry.get_mut() = PeerState::Backoff { timer, timer_deadline };
 
 					} else {
@@ -1324,7 +1324,7 @@ impl NetworkBehaviour for GenericProto {
 
 					if connections.is_empty() {
 						debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
-						self.peerset.dropped(set_id, peer_id.clone());
+						self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
 						let ban_dur = Uniform::new(5, 10).sample(&mut rand::thread_rng());
 
 						let delay_id = self.next_delay_id;
@@ -1345,7 +1345,7 @@ impl NetworkBehaviour for GenericProto {
 						matches!(s, ConnectionState::Opening | ConnectionState::Open(_)))
 					{
 						debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
-						self.peerset.dropped(set_id, peer_id.clone());
+						self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
 
 						*entry.get_mut() = PeerState::Disabled {
 							connections,
@@ -1396,7 +1396,7 @@ impl NetworkBehaviour for GenericProto {
 					st @ PeerState::Requested |
 					st @ PeerState::PendingRequest { .. } => {
 						debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", peer_id, set_id);
-						self.peerset.dropped(set_id, peer_id.clone());
+						self.peerset.dropped(set_id, peer_id.clone(), sc_peerset::DropReason::Unknown);
 
 						let now = Instant::now();
 						let ban_duration = match st {
@@ -1682,7 +1682,7 @@ impl NetworkBehaviour for GenericProto {
 							// List of open connections wasn't empty before but now it is.
 							if !connections.iter().any(|(_, s)| matches!(s, ConnectionState::Opening)) {
 								debug!(target: "sub-libp2p", "PSM <= Dropped({}, {:?})", source, set_id);
-								self.peerset.dropped(set_id, source.clone());
+								self.peerset.dropped(set_id, source.clone(), sc_peerset::DropReason::Refused);
 								*entry.into_mut() = PeerState::Disabled {
 									connections, backoff_until: None
 								};
@@ -1846,7 +1846,7 @@ impl NetworkBehaviour for GenericProto {
 							matches!(s, ConnectionState::Opening | ConnectionState::Open(_)))
 						{
 							debug!(target: "sub-libp2p", "PSM <= Dropped({:?})", source);
-							self.peerset.dropped(set_id, source.clone());
+							self.peerset.dropped(set_id, source.clone(), sc_peerset::DropReason::Refused);
 
 							*entry.into_mut() = PeerState::Disabled {
 								connections,

client/peerset/src/lib.rs

@@ -604,7 +604,7 @@ impl Peerset {
 	///
 	/// Must only be called after the PSM has either generated a `Connect` message with this
 	/// `PeerId`, or accepted an incoming connection with this `PeerId`.
-	pub fn dropped(&mut self, set_id: SetId, peer_id: PeerId) {
+	pub fn dropped(&mut self, set_id: SetId, peer_id: PeerId, reason: DropReason) {
 		// We want reputations to be up-to-date before adjusting them.
 		self.update_time();
 
@@ -620,6 +620,10 @@ impl Peerset {
 				error!(target: "peerset", "Received dropped() for non-connected node"),
 		}
 
+		if let DropReason::Refused = reason {
+			self.on_remove_from_peers_set(set_id, peer_id);
+		}
+
 		self.alloc_slots();
 	}
 
@@ -704,6 +708,17 @@ impl Stream for Peerset {
 	}
 }
 
+/// Reason for calling [`Peerset::dropped`].
+pub enum DropReason {
+	/// Substream or connection has been closed for an unknown reason.
+	Unknown,
+	/// Substream or connection has been explicitly refused by the target. In other words, the
+	/// peer doesn't actually belong to this set.
+	///
+	/// This has the side effect of calling [`PeersetHandle::remove_from_peers_set`].
+	Refused,
+}
+
 #[cfg(test)]
 mod tests {
 	use libp2p::PeerId;

client/peerset/tests/fuzz.rs

@@ -20,7 +20,7 @@ use futures::prelude::*;
 use libp2p::PeerId;
 use rand::distributions::{Distribution, Uniform, WeightedIndex};
 use rand::seq::IteratorRandom;
-use sc_peerset::{IncomingIndex, Message, Peerset, PeersetConfig, ReputationChange, SetConfig, SetId};
+use sc_peerset::{DropReason, IncomingIndex, Message, Peerset, PeersetConfig, ReputationChange, SetConfig, SetId};
 use std::{collections::HashMap, collections::HashSet, pin::Pin, task::Poll};
 
 #[test]
@@ -130,7 +130,7 @@ fn test_once() {
 				3 => {
 					if let Some(id) = connected_nodes.iter().choose(&mut rng).cloned() {
 						connected_nodes.remove(&id);
-						peerset.dropped(SetId::from(0), id);
+						peerset.dropped(SetId::from(0), id, DropReason::Unknown);
 					}
 				}