net: tear down peer session when writer dies (mempool_reorg flake)

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}"),

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);
+}

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