Merge remote-tracking branch 'grin_ardocrat/peers_fix' into grim

Author: ardocrat

p2p/src/peers.rs

@@ -61,6 +61,7 @@ impl Peers {
 	/// Adds the peer to our internal peer mapping. Note that the peer is still
 	/// returned so the server can run it.
 	pub fn add_connected(&self, peer: Arc<Peer>) -> Result<(), Error> {
+		let enough_outbound = self.enough_outbound_peers();
 		let peer_data: PeerData;
 		{
 			// Scope for peers vector lock - dont hold the peers lock while adding to lmdb
@@ -77,8 +78,10 @@ impl Peers {
 				ban_reason: ReasonForBan::None,
 				last_connected: Utc::now().timestamp(),
 			};
-			debug!("Adding newly connected peer {}.", peer_data.addr);
-			peers.insert(peer_data.addr, peer);
+			if !enough_outbound || !peer.info.is_outbound() {
+				debug!("Adding newly connected peer {}.", peer_data.addr);
+				peers.insert(peer_data.addr, peer);
+			}
 		}
 		debug!("Saving newly connected peer {}.", peer_data.addr);
 		if let Err(e) = self.save_peer(&peer_data) {
@@ -142,6 +145,7 @@ impl Peers {
 		}
 		false
 	}
+
 	/// Ban a peer, disconnecting it if we're currently connected
 	pub fn ban_peer(&self, peer_addr: PeerAddr, ban_reason: ReasonForBan) -> Result<(), Error> {
 		// Update the peer in peers db
@@ -261,6 +265,8 @@ impl Peers {
 						break;
 					}
 				};
+				// Mark peer as defunct after ping failure.
+				let _ = self.update_state(p.info.addr, State::Defunct);
 				p.stop();
 				peers.remove(&p.info.addr);
 			}
@@ -747,21 +753,24 @@ impl NetAdapter for Peers {
 		trace!("Received {} peer addrs, saving.", peer_addrs.len());
 		let mut to_save: Vec<PeerData> = Vec::new();
 		for pa in peer_addrs {
-			if let Ok(e) = self.exists_peer(pa) {
-				if e {
+			if let Ok(mut p) = self.get_peer(pa) {
+				if p.flags != State::Defunct {
 					continue;
 				}
+				p.flags = State::Unknown;
+				to_save.push(p);
+			} else {
+				let peer = PeerData {
+					addr: pa,
+					capabilities: Capabilities::UNKNOWN,
+					user_agent: "".to_string(),
+					flags: State::Unknown,
+					last_banned: 0,
+					ban_reason: ReasonForBan::None,
+					last_connected: 0,
+				};
+				to_save.push(peer);
 			}
-			let peer = PeerData {
-				addr: pa,
-				capabilities: Capabilities::UNKNOWN,
-				user_agent: "".to_string(),
-				flags: State::Healthy,
-				last_banned: 0,
-				ban_reason: ReasonForBan::None,
-				last_connected: Utc::now().timestamp(),
-			};
-			to_save.push(peer);
 		}
 		if let Err(e) = self.save_peers(to_save) {
 			error!("Could not save received peer addresses: {:?}", e);

p2p/src/serv.rs

@@ -187,6 +187,9 @@ impl Server {
 					&self.handshake,
 					self.peers.clone(),
 				)?;
+				if self.peers.enough_outbound_peers() {
+					peer.stop();
+				}
 				let peer = Arc::new(peer);
 				self.peers.add_connected(peer.clone())?;
 				Ok(peer)

p2p/src/store.rs

@@ -27,13 +27,14 @@ const STORE_SUBPATH: &str = "peers";
 
 const PEER_PREFIX: u8 = b'P';
 
-// Types of messages
+// Types of peers
 enum_from_primitive! {
 	#[derive(Debug, Clone, Copy, PartialEq, Serialize, Deserialize)]
 	pub enum State {
 		Healthy = 0,
 		Banned = 1,
 		Defunct = 2,
+		Unknown = 3,
 	}
 }
 

p2p/src/types.rs

@@ -183,10 +183,13 @@ impl<'de> Visitor<'de> for PeerAddrs {
 				Ok(ip) => peers.push(PeerAddr(ip)),
 				// If that fails it's probably a DNS record
 				Err(_) => {
-					let socket_addrs = entry.to_socket_addrs().map_err(|_| {
-						serde::de::Error::custom(format!("Unable to resolve DNS: {}", entry))
-					})?;
-					peers.append(&mut socket_addrs.map(PeerAddr).collect());
+					let socket_addrs: Result<std::vec::IntoIter<SocketAddr>, M::Error> =
+						entry.to_socket_addrs().map_err(|_| {
+							serde::de::Error::custom(format!("Unable to resolve DNS: {}", entry))
+						});
+					if let Ok(socket_addrs) = socket_addrs {
+						peers.append(&mut socket_addrs.map(PeerAddr).collect());
+					}
 				}
 			}
 		}