diff --git a/p2p/src/peers.rs b/p2p/src/peers.rs index adc24a4c..c3e32104 100644 --- a/p2p/src/peers.rs +++ b/p2p/src/peers.rs @@ -142,6 +142,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 @@ -178,6 +179,13 @@ impl Peers { } } + pub fn is_healthy(&self, peer_addr: PeerAddr) -> bool { + if let Ok(peer) = self.store.get_peer(peer_addr) { + return peer.flags == State::Healthy; + } + false + } + fn broadcast(&self, obj_name: &str, inner: F) -> u32 where F: Fn(&Peer) -> Result, @@ -261,6 +269,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); } @@ -702,21 +712,24 @@ impl NetAdapter for Peers { trace!("Received {} peer addrs, saving.", peer_addrs.len()); let mut to_save: Vec = 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 self.is_healthy(pa) || self.is_banned(pa) { 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); diff --git a/p2p/src/store.rs b/p2p/src/store.rs index 54dd6d2f..96909aab 100644 --- a/p2p/src/store.rs +++ b/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, } } diff --git a/p2p/src/types.rs b/p2p/src/types.rs index 07765496..76cc3fe7 100644 --- a/p2p/src/types.rs +++ b/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, 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()); + } } } } diff --git a/servers/src/grin/seed.rs b/servers/src/grin/seed.rs index b04f0afe..6c155e74 100644 --- a/servers/src/grin/seed.rs +++ b/servers/src/grin/seed.rs @@ -133,9 +133,10 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende // regularly check if we need to acquire more peers and if so, gets // them from db let mut total_count = 0; - let mut healthy_count = 0; let mut banned_count = 0; + let mut healthy = vec![]; let mut defuncts = vec![]; + let mut unknown = vec![]; for x in peers.all_peer_data().into_iter() { match x.flags { @@ -154,8 +155,9 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende banned_count += 1; } } - p2p::State::Healthy => healthy_count += 1, - p2p::State::Defunct => defuncts.push(x), + p2p::State::Healthy => healthy.push(x.addr), + p2p::State::Defunct => defuncts.push(x.addr), + p2p::State::Unknown => unknown.push(x.addr), } total_count += 1; } @@ -167,15 +169,16 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende debug!( "monitor_peers: on {}:{}, {} connected ({} most_work). \ - all {} = {} healthy + {} banned + {} defunct", + all {} = {} healthy + {} banned + {} defunct + {} unknown", config.host, config.port, peers_count, most_work_count, total_count, - healthy_count, + healthy.len(), banned_count, defuncts.len(), + unknown.len() ); // maintenance step first, clean up p2p server peers @@ -185,64 +188,74 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende config.clone(), ); - if peers.enough_outbound_peers() { - return; - } + if !peers.enough_outbound_peers() { + // loop over connected peers that can provide peer lists + // ask them for their list of peers + let mut connected_peers: Vec = vec![]; + for p in peers + .iter() + .with_capabilities(p2p::Capabilities::PEER_LIST) + .connected() + { + trace!( + "monitor_peers: {}:{} ask {} for more peers", + config.host, + config.port, + p.info.addr, + ); + let _ = p.send_peer_request(p2p::Capabilities::PEER_LIST); + connected_peers.push(p.info.addr) + } - // loop over connected peers that can provide peer lists - // ask them for their list of peers - let mut connected_peers: Vec = vec![]; - for p in peers - .iter() - .with_capabilities(p2p::Capabilities::PEER_LIST) - .connected() - { - trace!( - "monitor_peers: {}:{} ask {} for more peers", - config.host, - config.port, - p.info.addr, - ); - let _ = p.send_peer_request(p2p::Capabilities::PEER_LIST); - connected_peers.push(p.info.addr) - } - - // Attempt to connect to any preferred peers. - let peers_preferred = config.peers_preferred.unwrap_or(PeerAddrs::default()); - for p in peers_preferred { - if !connected_peers.is_empty() { - if !connected_peers.contains(&p) { + // Attempt to connect to any preferred peers. + let peers_preferred = config.peers_preferred.unwrap_or(PeerAddrs::default()); + for p in peers_preferred { + if !connected_peers.is_empty() { + if !connected_peers.contains(&p) { + let _ = tx.send(p); + } + } else { let _ = tx.send(p); } - } else { - let _ = tx.send(p); } } - // take a random defunct peer and mark it healthy: over a long enough period any - // peer will see another as defunct eventually, gives us a chance to retry - if let Some(peer) = defuncts.into_iter().choose(&mut thread_rng()) { - let _ = peers.update_state(peer.addr, p2p::State::Healthy); - } - // find some peers from our db // and queue them up for a connection attempt // intentionally make too many attempts (2x) as some (most?) will fail // as many nodes in our db are not publicly accessible + let mut new_peers = vec![]; let max_peer_attempts = 128; - let new_peers = peers.find_peers( - p2p::State::Healthy, - p2p::Capabilities::UNKNOWN, - max_peer_attempts as usize, - ); + // check random disconnected healthy peers. + for hpa in healthy + .iter() + .filter(|p| peers.get_connected_peer(**p).is_none()) + .choose_multiple(&mut thread_rng(), max_peer_attempts) + { + new_peers.push(hpa); + } + // check random unknown peers received from peer list request. + for upa in unknown + .iter() + .choose_multiple(&mut thread_rng(), max_peer_attempts) + { + new_peers.push(upa); + } + // always check random defunct peers. + for dpa in defuncts + .iter() + .choose_multiple(&mut thread_rng(), max_peer_attempts) + { + new_peers.push(dpa); + } // Only queue up connection attempts for candidate peers where we // are confident we do not yet know about this peer. // The call to is_known() may fail due to contention on the peers map. // Do not attempt any connection where is_known() fails for any reason. - for p in new_peers { - if let Ok(false) = peers.is_known(p.addr) { - tx.send(p.addr).unwrap(); + for pa in new_peers { + if let Ok(false) = peers.is_known(*pa) { + tx.send(*pa).unwrap(); } } } @@ -275,7 +288,7 @@ fn connect_to_seeds_and_peers( // check if we have some peers in db // look for peers that are able to give us other peers (via PEER_LIST capability) - let peers = peers.find_peers(p2p::State::Healthy, p2p::Capabilities::PEER_LIST, 100); + let peers = peers.find_peers(p2p::State::Healthy, p2p::Capabilities::PEER_LIST, 128); // if so, get their addresses, otherwise use our seeds let peer_addrs = if peers.len() > 3 { @@ -307,19 +320,14 @@ fn listen_for_addrs( ) { // Pull everything currently on the queue off the queue. // Does not block so addrs may be empty. - // We will take(max_peers) from this later but we want to drain the rx queue + // We will take(max_peers) from this later, but we want to drain the rx queue // here to prevent it backing up. let addrs: Vec = rx.try_iter().collect(); - // If we have a healthy number of outbound peers then we are done here. - if peers.enough_outbound_peers() { - return; - } - // Note: We drained the rx queue earlier to keep it under control. // Even if there are many addresses to try we will only try a bounded number of them for safety. let connect_min_interval = 30; - let max_outbound_attempts = 128; + let max_outbound_attempts = 128 * 3; for addr in addrs.into_iter().take(max_outbound_attempts) { // ignore the duplicate connecting to same peer within 30 seconds let now = Utc::now(); @@ -349,7 +357,6 @@ fn listen_for_addrs( if p.info.capabilities.contains(p2p::Capabilities::PEER_LIST) { let _ = p.send_peer_request(p2p::Capabilities::PEER_LIST); } - let _ = peers_c.update_state(addr, p2p::State::Healthy); } Err(_) => { let _ = peers_c.update_state(addr, p2p::State::Defunct);