From 72f68ebc784fb17a36676ba7857934d9112bfeca Mon Sep 17 00:00:00 2001 From: "rbitcoin-grok[bot]" Date: Sun, 30 Aug 2026 07:43:05 -0700 Subject: [PATCH] net: tear down peer session when writer dies (mempool_reorg) After tip sync, disconnect_nodes still flaked waiting for the far side's getpeerinfo: a write failure or BIP324 close mapped oddly could leave the session in getpeerinfo past 5s. Select on writer-task completion, treat any socket Io as a clean peer-gone exit, and map ConnectionAborted through as Io. --- crates/rbitcoin-net/src/peer.rs | 31 ++++---- crates/rbitcoin-net/src/peer_tests.rs | 107 ++++++++++++++++++++++++++ crates/rbitcoin-net/src/v2.rs | 11 ++- 3 files changed, 131 insertions(+), 18 deletions(-) diff --git a/crates/rbitcoin-net/src/peer.rs b/crates/rbitcoin-net/src/peer.rs index c1581673..c2962463 100644 --- a/crates/rbitcoin-net/src/peer.rs +++ b/crates/rbitcoin-net/src/peer.rs @@ -679,7 +679,7 @@ pub async fn peer_session_with( } let writer_session = meta.session.clone(); - let writer_task = tokio::spawn(async move { + let mut writer_task = tokio::spawn(async move { while let Some(msg) = out_rx.recv().await { let full = matches!( msg, @@ -736,6 +736,12 @@ pub async fn peer_session_with( } tokio::select! { biased; + // Peer half-close / write failure: tear down so getpeerinfo + // clears without waiting on a stuck read/decode arm. + writer_done = &mut writer_task => { + let _ = writer_done; + return Ok(()); + } _ = tokio::time::sleep(Duration::from_millis(50)), if session.is_some() => { if tx_announce_rx.is_none() { tx_announce_rx = hub.mempool().map(|m| m.subscribe_announces()); @@ -967,17 +973,9 @@ pub async fn peer_session_with( frame = read_v2_frame(&mut reader, magic) => { let frame = match frame { Ok(f) => f, - Err(NetError::Io(e)) - if matches!( - e.kind(), - std::io::ErrorKind::UnexpectedEof - | std::io::ErrorKind::ConnectionReset - | std::io::ErrorKind::BrokenPipe - | std::io::ErrorKind::ConnectionAborted - ) => - { - return Ok(()); - } + // Any socket Io means the peer is gone — exit cleanly so + // unregister runs inside the Core disconnect_nodes 5s wait. + Err(NetError::Io(_)) => return Ok(()), Err(NetError::MessageTooLarge(n)) => { ban_score = ban_score.saturating_add(OVERSIZE_BAN_SCORE); rbitcoin_log::warn!( @@ -1059,8 +1057,13 @@ pub async fn peer_session_with( .await; drop(out_tx); - writer_task.abort(); - let _ = writer_task.await; + // If select! already joined the writer, do not await again (would pend). + if writer_task.is_finished() { + drop(writer_task); + } else { + writer_task.abort(); + let _ = writer_task.await; + } match &result { Ok(()) => rbitcoin_log::debug!("p2p: session {peer_s} closed"), Err(e) => rbitcoin_log::warn!("p2p: session {peer_s} ended: {e}"), diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index b203e338..3a959d8b 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -5667,3 +5667,110 @@ async fn disconnect_clears_far_side_getpeerinfo_within_5s() { nb.shutdown().await; let _ = std::fs::remove_dir_all(dir); } + +/// `mempool_reorg` disconnect_nodes after generate+sync — far side must still +/// clear within 5s even if the session was just busy accepting tip blocks. +#[tokio::test(flavor = "multi_thread", worker_threads = 2)] +async fn disconnect_after_tip_sync_clears_far_side_within_5s() { + use crate::P2PNode; + use bitcoin::ScriptBuf; + use std::time::Duration; + + let n = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let dir = std::env::temp_dir().join(format!("rbitcoin-disc-tip-{n}")); + std::fs::create_dir_all(dir.join("a")).unwrap(); + std::fs::create_dir_all(dir.join("b")).unwrap(); + let qa = Query::open_or_create(dir.join("a/store")).unwrap(); + let qb = Query::open_or_create(dir.join("b/store")).unwrap(); + let params = ChainParams::regtest(); + let mut na = P2PNode::start_with_agent( + "127.0.0.1:0".parse().unwrap(), + qa, + params.clone(), + Milestone::NONE, + "/rbitcoin:0.1.0(testnode0)/".into(), + crate::DEFAULT_MAX_INBOUND, + ) + .await + .unwrap(); + let nb = P2PNode::start_with_agent( + "127.0.0.1:0".parse().unwrap(), + qb, + params, + Milestone::NONE, + "/rbitcoin:0.1.0(testnode1)/".into(), + crate::DEFAULT_MAX_INBOUND, + ) + .await + .unwrap(); + + na.follow_from(nb.local_addr).await.unwrap(); + let mut linked = false; + for _ in 0..100 { + let a_sees = na + .peers + .snapshot() + .iter() + .any(|p| p.subver.contains("testnode1")); + let b_sees = nb + .peers + .snapshot() + .iter() + .any(|p| p.subver.contains("testnode0")); + if a_sees && b_sees { + linked = true; + break; + } + tokio::time::sleep(Duration::from_millis(20)).await; + } + assert!(linked, "both sides must list each other before generate"); + + const BURST: u32 = 3; + na.hub + .generate_to_script(BURST, ScriptBuf::from_bytes(vec![0x51]), vec![]) + .unwrap(); + let want = na.tip_height().unwrap(); + let mut synced = false; + for _ in 0..200 { + if nb.tip_height().unwrap_or(0) >= want { + synced = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!(synced, "far side must sync tip before disconnect"); + + let peer_id = na + .peers + .snapshot() + .into_iter() + .find(|p| p.subver.contains("testnode1")) + .map(|p| p.id) + .expect("outbound peer id"); + assert!(na.peers.disconnect_id(peer_id)); + + let mut far_clear = false; + for _ in 0..100 { + if !nb + .peers + .snapshot() + .iter() + .any(|p| p.subver.contains("testnode0")) + { + far_clear = true; + break; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + assert!( + far_clear, + "far side getpeerinfo must drop us within 5s after tip sync (mempool_reorg)" + ); + + na.shutdown().await; + nb.shutdown().await; + let _ = std::fs::remove_dir_all(dir); +} diff --git a/crates/rbitcoin-net/src/v2.rs b/crates/rbitcoin-net/src/v2.rs index e9e94d6b..8e7fb630 100644 --- a/crates/rbitcoin-net/src/v2.rs +++ b/crates/rbitcoin-net/src/v2.rs @@ -431,10 +431,13 @@ fn map_protocol_error(e: ProtocolError) -> NetError { // completed v2 then dropped us. Prefer the IO detail when present so // logs are not all "does not speak BIP324 v2". ProtocolError::Io(io, ProtocolFailureSuggestion::RetryV1) => { - if io.kind() == std::io::ErrorKind::UnexpectedEof - || io.kind() == std::io::ErrorKind::ConnectionReset - || io.kind() == std::io::ErrorKind::BrokenPipe - { + if matches!( + io.kind(), + std::io::ErrorKind::UnexpectedEof + | std::io::ErrorKind::ConnectionReset + | std::io::ErrorKind::BrokenPipe + | std::io::ErrorKind::ConnectionAborted + ) { NetError::Io(io) } else { NetError::V1Peer