diff --git a/p2p/src/types.rs b/p2p/src/types.rs index 5d90223e8c..3f7fd69d0e 100644 --- a/p2p/src/types.rs +++ b/p2p/src/types.rs @@ -671,6 +671,25 @@ impl PeerInfo { live_info.total_difficulty = total_difficulty; live_info.last_seen = Utc::now() } + + /// Update height/difficulty only when the peer advertises *more* work + /// (higher total_difficulty, or same difficulty at a greater height). + /// Used when a peer sends us a header/block/compact block so we do not wait + /// for the next ping/pong (up to ~10s) to reflect their chain tip. + pub fn update_if_better(&self, height: u64, total_difficulty: Difficulty) { + let mut live_info = self.live_info.write(); + let better = total_difficulty > live_info.total_difficulty + || (total_difficulty == live_info.total_difficulty && height > live_info.height); + if !better { + return; + } + if total_difficulty != live_info.total_difficulty { + live_info.stuck_detector = Utc::now(); + } + live_info.height = height; + live_info.total_difficulty = total_difficulty; + live_info.last_seen = Utc::now() + } } /// Flatten out a PeerInfo and nested PeerLiveInfo (taking a read lock on it) @@ -903,6 +922,8 @@ pub trait NetAdapter: ChainAdapter { fn peer_addrs_received(&self, _: Vec); /// Heard total_difficulty from a connected peer (via ping/pong). + /// Header/block/compact-block receipts also update peer live info directly + /// via [`PeerInfo::update_if_better`] so tip tracking is not delayed until ping. fn peer_difficulty(&self, _: PeerAddr, _: Difficulty, _: u64); /// Is this peer currently banned? @@ -924,3 +945,56 @@ pub struct AttachmentUpdate { pub left: usize, pub meta: Arc, } + +#[cfg(test)] +mod test { + use super::*; + use crate::core::ser::ProtocolVersion; + use std::net::{IpAddr, Ipv4Addr, SocketAddr}; + + fn test_peer_info(difficulty: Difficulty) -> PeerInfo { + PeerInfo { + capabilities: Capabilities::default(), + user_agent: "test".into(), + version: ProtocolVersion(1), + addr: PeerAddr(SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 3414)), + direction: Direction::Outbound, + live_info: Arc::new(RwLock::new(PeerLiveInfo::new(difficulty))), + } + } + + #[test] + fn update_if_better_increases_work() { + let peer = test_peer_info(Difficulty::from_num(10)); + assert_eq!(peer.total_difficulty(), Difficulty::from_num(10)); + assert_eq!(peer.height(), 0); + + // lower difficulty is ignored + peer.update_if_better(5, Difficulty::from_num(5)); + assert_eq!(peer.total_difficulty(), Difficulty::from_num(10)); + assert_eq!(peer.height(), 0); + + // equal difficulty, equal height ignored + peer.update_if_better(0, Difficulty::from_num(10)); + assert_eq!(peer.height(), 0); + + // equal difficulty, higher height accepted + peer.update_if_better(3, Difficulty::from_num(10)); + assert_eq!(peer.total_difficulty(), Difficulty::from_num(10)); + assert_eq!(peer.height(), 3); + + // equal difficulty, lower height ignored + peer.update_if_better(1, Difficulty::from_num(10)); + assert_eq!(peer.height(), 3); + + // higher difficulty accepted + peer.update_if_better(4, Difficulty::from_num(20)); + assert_eq!(peer.total_difficulty(), Difficulty::from_num(20)); + assert_eq!(peer.height(), 4); + + // absolute update still overwrites (ping/pong) + peer.update(2, Difficulty::from_num(15)); + assert_eq!(peer.total_difficulty(), Difficulty::from_num(15)); + assert_eq!(peer.height(), 2); + } +} diff --git a/servers/src/common/adapters.rs b/servers/src/common/adapters.rs index 0513d7ecca..b72b53ff64 100644 --- a/servers/src/common/adapters.rs +++ b/servers/src/common/adapters.rs @@ -165,7 +165,9 @@ where peer_info: &PeerInfo, opts: Options, ) -> Result { + // Peer is advertising this work even if we already know the block. if self.chain().is_known(&b.header).is_err() { + Self::update_peer_from_header(peer_info, &b.header); return Ok(true); } @@ -187,7 +189,9 @@ where peer_info: &PeerInfo, ) -> Result { // No need to process this compact block if we have previously accepted the _full block_. + // Still refresh peer difficulty/height from the header they sent. if self.chain().is_known(&cb.header).is_err() { + Self::update_peer_from_header(peer_info, &cb.header); return Ok(true); } @@ -229,8 +233,13 @@ where // check at least the header is valid before hydrating if let Err(e) = self.chain().process_block_header(&cb.header, Options::NONE) { debug!("Invalid compact block header {}: {:?}", cb_hash, e); + // Already-known / non-ban errors: still refresh peer work if we have the header. + if !e.is_bad_data() && self.chain().get_block_header(&cb.header.hash()).is_ok() { + Self::update_peer_from_header(peer_info, &cb.header); + } return Ok(!e.is_bad_data()); } + Self::update_peer_from_header(peer_info, &cb.header); let (txs, missing_short_ids) = { self.tx_pool @@ -291,7 +300,9 @@ where fn header_received(&self, bh: BlockHeader, peer_info: &PeerInfo) -> Result { // No need to process this header if we have previously accepted the _full block_. + // Still update peer live info — sending the header advertises that work. if self.chain().block_exists(bh.hash())? { + Self::update_peer_from_header(peer_info, &bh); return Ok(true); } if !self.sync_state.is_syncing() { @@ -309,6 +320,11 @@ where if e.is_bad_data() { return Ok(false); } else { + // Header may still be known on the header chain (e.g. already seen during sync). + // Refresh peer work when the header is not bad data. + if self.chain().get_block_header(&bh.hash()).is_ok() { + Self::update_peer_from_header(peer_info, &bh); + } // we got an error when trying to process the block header // but nothing serious enough to need to ban the peer upstream return Err(e); @@ -316,6 +332,7 @@ where } // we have successfully processed a block header + Self::update_peer_from_header(peer_info, &bh); // so we can go request the block itself self.request_compact_block(&bh, peer_info); @@ -348,7 +365,13 @@ where } }; - self.process_header_batch(bhs, sync_head) + let accepted = self.process_header_batch(bhs, sync_head)?; + if accepted { + if let Some(best) = bhs.iter().max_by_key(|h| h.total_difficulty()) { + Self::update_peer_from_header(peer_info, best); + } + } + Ok(accepted) } fn locate_headers(&self, locator: &[Hash]) -> Result, chain::Error> { @@ -1073,6 +1096,13 @@ where .reject_pihd_header_segment_from(entry.id, entry.peer_info.addr.0); return Ok(Some(entry.peer_info)); } + if let Some(best) = entry + .headers + .iter() + .max_by_key(|h| h.total_difficulty()) + { + Self::update_peer_from_header(&entry.peer_info, best); + } self.sync_state .remove_pihd_header_segment(entry.id, entry.peer_info.addr.0); } @@ -1154,6 +1184,12 @@ where None } + /// Refresh peer height/difficulty from a header they sent us (header-first + /// gossip, compact block, or full block). Only increases advertised work. + fn update_peer_from_header(peer_info: &PeerInfo, header: &BlockHeader) { + peer_info.update_if_better(header.height, header.total_difficulty()); + } + // pushing the new block through the chain pipeline // remembering to reset the head if we have a bad block fn process_block( @@ -1174,10 +1210,12 @@ where } let bhash = b.hash(); + let header = b.header.clone(); let previous = self.chain().get_previous_header(&b.header); match self.chain().process_block(b, opts) { Ok(_) => { + Self::update_peer_from_header(peer_info, &header); self.validate_chain(bhash); self.check_compact(); Ok(true) @@ -1189,6 +1227,8 @@ where Err(e) => { match e { chain::Error::Orphan => { + // Peer has this block (orphan to us); still advertise their work. + Self::update_peer_from_header(peer_info, &header); if let Ok(previous) = previous { // make sure we did not miss the parent block if !self.chain().is_orphan(&previous.hash()) @@ -1202,6 +1242,10 @@ where } _ => { debug!("process_block: block {} refused by chain: {}", bhash, e); + // Already known / unfit: refresh peer only if we have this header. + if self.chain().get_block_header(&header.hash()).is_ok() { + Self::update_peer_from_header(peer_info, &header); + } Ok(true) } }