From ab9715d9a92fe7ebe257d469f677dd2c0c9ba8a1 Mon Sep 17 00:00:00 2001 From: ardocrat Date: Sat, 4 Apr 2026 01:58:35 +0300 Subject: [PATCH 1/2] peer: unknown state for new peers, check peers state on every monitor (128 healthy non-connected + 128 defuncts + 128 unknown), mark peer as defunct when ping not passed, do not crash on toml parse with dns failure --- p2p/src/peers.rs | 37 ++++++++---- p2p/src/store.rs | 3 +- p2p/src/types.rs | 11 ++-- servers/src/grin/seed.rs | 119 +++++++++++++++++++++------------------ 4 files changed, 97 insertions(+), 73 deletions(-) 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); From 9585c02135e773e361e714b1408f39d71e7a740e Mon Sep 17 00:00:00 2001 From: ardocrat Date: Mon, 6 Apr 2026 01:44:22 +0300 Subject: [PATCH 2/2] p2p: cleanup before selection at monitor, add outbound to connected list only when there is not enough peers + disconnect extra peer immediately, reconnect to seeds at monitor to avoid stuck, update only defunct state to unknown when received existing peer address --- p2p/src/peers.rs | 16 +++--- p2p/src/serv.rs | 3 ++ servers/src/grin/seed.rs | 103 +++++++++++++++++++++---------------- servers/src/grin/server.rs | 18 ------- 4 files changed, 69 insertions(+), 71 deletions(-) diff --git a/p2p/src/peers.rs b/p2p/src/peers.rs index c3e32104..3448c6dd 100644 --- a/p2p/src/peers.rs +++ b/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) -> 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) { @@ -179,13 +182,6 @@ 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, @@ -713,7 +709,7 @@ impl NetAdapter for Peers { let mut to_save: Vec = Vec::new(); for pa in peer_addrs { if let Ok(mut p) = self.get_peer(pa) { - if self.is_healthy(pa) || self.is_banned(pa) { + if p.flags != State::Defunct { continue; } p.flags = State::Unknown; diff --git a/p2p/src/serv.rs b/p2p/src/serv.rs index c31c549c..2de9c1f8 100644 --- a/p2p/src/serv.rs +++ b/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) diff --git a/servers/src/grin/seed.rs b/servers/src/grin/seed.rs index 6c155e74..7af7fe8b 100644 --- a/servers/src/grin/seed.rs +++ b/servers/src/grin/seed.rs @@ -53,7 +53,6 @@ pub const TESTNET_DNS_SEEDS: &[&str] = &[ pub fn connect_and_monitor( p2p_server: Arc, - seed_list: Box Vec + Send>, config: P2PConfig, stop_state: Arc, ) -> std::io::Result> { @@ -67,7 +66,7 @@ pub fn connect_and_monitor( let (tx, rx) = mpsc::channel(); // check seeds first - connect_to_seeds_and_peers(peers.clone(), tx.clone(), seed_list, config); + connect_to_seeds_and_peers(peers.clone(), tx.clone(), config); let mut prev = DateTime::::MIN_UTC; let mut prev_expire_check = DateTime::::MIN_UTC; @@ -130,6 +129,13 @@ pub fn connect_and_monitor( } fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sender) { + // maintenance step first, clean up p2p server peers + peers.clean_peers( + config.peer_max_inbound_count() as usize, + config.peer_max_outbound_count() as usize, + config.clone(), + ); + // regularly check if we need to acquire more peers and if so, gets // them from db let mut total_count = 0; @@ -181,12 +187,12 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende unknown.len() ); - // maintenance step first, clean up p2p server peers - peers.clean_peers( - config.peer_max_inbound_count() as usize, - config.peer_max_outbound_count() as usize, - config.clone(), - ); + // Connect to seeds again if there is no peers at database, + // helps to avoid stuck when 1st request to seed list was failed. + if total_count == 0 { + connect_to_seeds_and_peers(peers.clone(), tx.clone(), config); + return; + } if !peers.enough_outbound_peers() { // loop over connected peers that can provide peer lists @@ -208,14 +214,15 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende } // Attempt to connect to any preferred peers. - let peers_preferred = config.peers_preferred.unwrap_or(PeerAddrs::default()); - for p in peers_preferred { + let default_peers = PeerAddrs::default(); + let peers_preferred = config.peers_preferred.as_ref().unwrap_or(&default_peers); + for p in peers_preferred.peers.iter() { if !connected_peers.is_empty() { if !connected_peers.contains(&p) { - let _ = tx.send(p); + let _ = tx.send(*p); } } else { - let _ = tx.send(p); + let _ = tx.send(*p); } } } @@ -265,10 +272,10 @@ fn monitor_peers(peers: Arc, config: p2p::P2PConfig, tx: mpsc::Sende fn connect_to_seeds_and_peers( peers: Arc, tx: mpsc::Sender, - seed_list: Box Vec>, config: P2PConfig, ) { - let peers_deny = config.peers_deny.unwrap_or(PeerAddrs::default()); + let default_peers = PeerAddrs::default(); + let peers_deny = config.peers_deny.as_ref().unwrap_or(&default_peers); // If "peers_allow" is explicitly configured then just use this list // remembering to filter out "peers_deny". @@ -280,7 +287,7 @@ fn connect_to_seeds_and_peers( } // Always try our "peers_preferred" remembering to filter out "peers_deny". - if let Some(peers) = config.peers_preferred { + if let Some(peers) = config.peers_preferred.as_ref() { for addr in peers.difference(peers_deny.as_slice()) { let _ = tx.send(addr); } @@ -294,7 +301,7 @@ fn connect_to_seeds_and_peers( let peer_addrs = if peers.len() > 3 { peers.iter().map(|p| p.addr).collect::>() } else { - seed_list() + seed_list(&config) }; if peer_addrs.is_empty() { @@ -380,27 +387,43 @@ fn listen_for_addrs( } } -pub fn default_dns_seeds() -> Box Vec + Send> { - Box::new(|| { - let net_seeds = if global::is_testnet() { - TESTNET_DNS_SEEDS - } else { - MAINNET_DNS_SEEDS - }; - resolve_dns_to_addrs( - &net_seeds - .iter() - .map(|s| { - s.to_string() - + if global::is_testnet() { - ":13414" - } else { - ":3414" - } - }) - .collect(), - ) - }) +fn seed_list(config: &P2PConfig) -> Vec { + match config.seeding_type { + p2p::Seeding::None => { + warn!("No seed configured, will stay solo until connected to"); + vec![] + } + p2p::Seeding::List => match &config.seeds { + Some(seeds) => seeds.peers.clone(), + None => { + error!("Seeds must be configured for seeding type List"); + vec![] + } + }, + p2p::Seeding::DNSSeed => default_dns_seeds(), + _ => vec![], + } +} + +fn default_dns_seeds() -> Vec { + let net_seeds = if global::is_testnet() { + TESTNET_DNS_SEEDS + } else { + MAINNET_DNS_SEEDS + }; + resolve_dns_to_addrs( + &net_seeds + .iter() + .map(|s| { + s.to_string() + + if global::is_testnet() { + ":13414" + } else { + ":3414" + } + }) + .collect(), + ) } /// Convenience function to resolve dns addresses from DNS records @@ -421,9 +444,3 @@ pub fn resolve_dns_to_addrs(dns_records: &Vec) -> Vec { debug!("Resolved addresses: {:?}", addresses); addresses } - -/// Convenience function when the seed list is immediately known. Mostly used -/// for tests. -pub fn predefined_seeds(addrs: Vec) -> Box Vec + Send> { - Box::new(move || addrs.clone()) -} diff --git a/servers/src/grin/server.rs b/servers/src/grin/server.rs index 5cfd92c1..d3510fe2 100644 --- a/servers/src/grin/server.rs +++ b/servers/src/grin/server.rs @@ -237,26 +237,8 @@ impl Server { let mut connect_thread = None; if config.p2p_config.seeding_type != p2p::Seeding::Programmatic { - let seed_list = match config.p2p_config.seeding_type { - p2p::Seeding::None => { - warn!("No seed configured, will stay solo until connected to"); - seed::predefined_seeds(vec![]) - } - p2p::Seeding::List => match &config.p2p_config.seeds { - Some(seeds) => seed::predefined_seeds(seeds.peers.clone()), - None => { - return Err(Error::Configuration( - "Seeds must be configured for seeding type List".to_owned(), - )); - } - }, - p2p::Seeding::DNSSeed => seed::default_dns_seeds(), - _ => unreachable!(), - }; - connect_thread = Some(seed::connect_and_monitor( p2p_server.clone(), - seed_list, config.p2p_config.clone(), stop_state.clone(), )?);