net: kill the 2026-10-17 missed mutants

changelog.d/overnight-39004835061.md

@@ -0,0 +1,3 @@
+Changed
+
+- Overnight missed mutants from 2026-10-17 are gone or pinned. The startup Class A "filled N" line is gone because the body-queue rehydrate already logs a non-zero fill.

crates/rbitcoin-net/src/compact.rs

@@ -192,9 +192,8 @@ pub(crate) fn reconstruct<T: Borrow<Transaction>>(
                 }
                 *slot = Some(cands[0].borrow().clone());
             }
-            Some(cands) if cands.len() > 1 => {
-                missing.push(abs as u64);
-            }
+            // One candidate fills the slot. Every other count, including a
+            // short-id collision, is missing for getblocktxn.
             _ => missing.push(abs as u64),
         }
     }
@@ -1034,6 +1033,20 @@ mod tests {
         assert_eq!(classify_v2_cmpct_peer(&ping), CmpctPeerFrame::Ping(7));
         let pong = crate::v2::encode_v2_contents(NetworkMessage::Pong(7)).unwrap();
         assert_eq!(classify_v2_cmpct_peer(&pong), CmpctPeerFrame::Pong(7));
+        let block_hash = BlockHash::from_byte_array([7; 32]);
+        let get = crate::v2::encode_v2_contents(NetworkMessage::GetBlockTxn(
+            bitcoin::p2p::message_compact_blocks::GetBlockTxn {
+                txs_request: bitcoin::bip152::BlockTransactionsRequest {
+                    block_hash,
+                    indexes: vec![1, 4],
+                },
+            },
+        ))
+        .unwrap();
+        assert_eq!(
+            classify_v2_cmpct_peer(&get),
+            CmpctPeerFrame::GetBlockTxn(vec![1, 4])
+        );
         assert_eq!(classify_v2_cmpct_peer(&[]), CmpctPeerFrame::Other);
     }
 
@@ -1262,7 +1275,13 @@ mod tests {
         assert_eq!(stats.fetched_n, 0);
         assert_eq!(stats.fetched_bytes, 0);
         assert_eq!(stats.mempool_n, 2);
+        assert_eq!(
+            stats.mempool_bytes,
+            tx_wire_len(&b1) + tx_wire_len(&b2),
+            "mempool fill bytes are the sum of the filled transactions"
+        );
         assert_eq!(stats.prefill_n, 1);
+        assert_eq!(stats.prefill_bytes, tx_wire_len(&block.txdata[0]));
         assert!(!stats.getdata);
         let line = stats.to_string();
         assert!(
@@ -1344,10 +1363,34 @@ mod tests {
         };
         let stats = reconstruct_stats(&hsi, &recon, &fill, &[]);
         assert_eq!(stats.extra_n, 1);
+        assert_eq!(stats.extra_bytes, tx_wire_len(&b1));
         assert_eq!(stats.mempool_n, 0);
         assert_eq!(stats.fetched_n, 0);
     }
 
+    #[test]
+    fn cmpct_reconstruct_stats_orphan_bytes_sum() {
+        let b1 = spend(15);
+        let block = sealed_block(vec![coinbase(), b1.clone()]);
+        let hsi = HeaderAndShortIds::from_block(&block, 8, 2, &[]).unwrap();
+        let recon = try_reconstruct(
+            &hsi,
+            &shortid_map_from_txs(&block.header, hsi.nonce, 2, [&b1]),
+            2,
+        )
+        .expect("orphan fill");
+        let fill = CmpctFillSets {
+            mempool: Default::default(),
+            extra: Default::default(),
+            orphan: [b1.compute_wtxid()].into_iter().collect(),
+        };
+        let stats = reconstruct_stats(&hsi, &recon, &fill, &[]);
+        assert_eq!(stats.orphan_n, 1);
+        assert_eq!(stats.orphan_bytes, tx_wire_len(&b1));
+        assert_eq!(stats.extra_n, 0);
+        assert_eq!(stats.mempool_n, 0);
+    }
+
     #[test]
     fn cmpct_reconstruct_stats_getdata_line() {
         let hash = BlockHash::from_byte_array([0xab; 32]);
@@ -1454,4 +1497,28 @@ mod tests {
             "missed mempool read is not an empty fill"
         );
     }
+
+    #[test]
+    fn prefill_indexes_keeps_a_tx_that_fills_the_budget_exactly() {
+        let mut pad = EXTRA_PREFILL_BUDGET;
+        let tx = loop {
+            let candidate = spend_padded(40, pad);
+            let sz = tx_wire_len(&candidate);
+            if sz == EXTRA_PREFILL_BUDGET {
+                break candidate;
+            }
+            if sz > EXTRA_PREFILL_BUDGET {
+                pad = pad.saturating_sub(sz - EXTRA_PREFILL_BUDGET);
+            } else {
+                pad += EXTRA_PREFILL_BUDGET - sz;
+            }
+        };
+        assert_eq!(tx_wire_len(&tx), EXTRA_PREFILL_BUDGET);
+        let block = sealed_block(vec![coinbase(), tx]);
+        assert_eq!(
+            prefill_indexes(&block, &CmpctFillSets::default()),
+            vec![0, 1],
+            "a transaction that lands on the extra-prefill budget is packed"
+        );
+    }
 }

crates/rbitcoin-net/src/eviction.rs

@@ -106,9 +106,6 @@ fn remove_first_k(cands: &mut Vec<InboundEvictCandidate>, k: usize) {
 
 /// Protect up to `k` peers from the largest keyed netgroups (Core netgroup protect).
 fn protect_by_netgroup(cands: &mut Vec<InboundEvictCandidate>, k: usize) {
-    if k == 0 || cands.is_empty() {
-        return;
-    }
     use std::collections::HashMap;
     let mut counts: HashMap<u64, usize> = HashMap::new();
     for c in cands.iter() {
@@ -262,6 +259,35 @@ mod tests {
         assert!(select_inbound_eviction(cands).is_none());
     }
 
+    #[test]
+    fn protect_by_netgroup_keeps_one_peer_from_each_largest_group() {
+        let mut untouched = vec![cand(1, 1, None, 0, 0)];
+        protect_by_netgroup(&mut untouched, 0);
+        assert_eq!(untouched.len(), 1, "k = 0 protects nobody");
+        let mut empty = Vec::new();
+        protect_by_netgroup(&mut empty, 4);
+        assert!(empty.is_empty());
+
+        let group = |id, netgroup| InboundEvictCandidate {
+            netgroup,
+            ..cand(id, id, None, 0, 0)
+        };
+        // Group 10 has two peers. Group 1 has one. k = 1 protects the lowest
+        // id in the larger group and leaves the other two.
+        let mut cands = vec![group(1, 10), group(2, 10), group(3, 1)];
+        protect_by_netgroup(&mut cands, 1);
+        let ids: Vec<_> = cands.iter().map(|c| c.id).collect();
+        assert_eq!(ids, vec![2, 3]);
+
+        let mut capped = vec![group(1, 10), group(3, 1)];
+        protect_by_netgroup(&mut capped, 1);
+        assert_eq!(capped.iter().map(|c| c.id).collect::<Vec<_>>(), vec![1]);
+
+        let mut mismatched = vec![group(8, 4), group(9, 5)];
+        protect_by_netgroup(&mut mismatched, 1);
+        assert_eq!(mismatched.iter().map(|c| c.id).collect::<Vec<_>>(), vec![9]);
+    }
+
     #[test]
     fn eviction_netgroup_ipv4_slash16() {
         let a: std::net::SocketAddr = "1.2.3.4:1".parse().unwrap();

crates/rbitcoin-net/src/i2p_sam.rs

@@ -421,7 +421,8 @@ fn decode_i2p_b64(s: &str) -> Result<Vec<u8>, NetError> {
                 )));
             }
         };
-        acc = (acc << 6) | v;
+        // `v` is six bits, so it does not overlap the shifted accumulator.
+        acc = (acc << 6) + v;
         n += 6;
         if n >= 8 {
             n -= 8;
@@ -1144,4 +1145,175 @@ mod tests {
         assert!(i2p_addr_from_sam_b64("!!!").is_err());
         assert!(i2p_addr_from_sam_b64(&i2p_b64_encode(&[1, 2, 3])).is_err());
     }
+
+    fn enc(msg: &str) -> NetError {
+        NetError::Encode(msg.to_string())
+    }
+
+    #[test]
+    fn sam_errors_name_the_socket_and_the_reply_that_retry() {
+        let io = NetError::Io(std::io::Error::other("sam io"));
+        assert!(session_dead(&io));
+        assert!(session_dead(&NetError::Disconnected));
+        assert!(sam_retry(&io));
+        assert!(sam_retry(&NetError::Disconnected));
+        assert!(stream_retry(&io));
+        assert!(stream_retry(&NetError::Disconnected));
+        assert!(session_dead(&enc("STREAM CONNECT: refused")));
+        assert!(!session_dead(&enc("i2p sam dial: connection refused")));
+        for needle in [
+            "connection reset",
+            "connection closed",
+            "i2p sam connect",
+            "early eof",
+            "unexpected eof",
+            "broken pipe",
+        ] {
+            assert!(sam_retry(&enc(needle)), "{needle}");
+        }
+        assert!(!sam_retry(&enc("i2p sam stream: RESULT=I2P_ERROR")));
+        for needle in [
+            "unexpected eof",
+            "broken pipe",
+            "connection reset",
+            "connection closed",
+            "cant_reach",
+            "early eof",
+            "timeout",
+        ] {
+            assert!(stream_retry(&enc(needle)), "{needle}");
+        }
+        assert!(!stream_retry(&enc("i2p sam stream: RESULT=I2P_ERROR")));
+    }
+
+    #[test]
+    fn i2p_b64_round_trips_digits_dash_and_tilde() {
+        for raw in [
+            vec![0, 0, 52],
+            vec![0, 0, 62],
+            vec![0, 0, 63],
+            vec![1, 2, 3, 4, 5],
+        ] {
+            let text = i2p_b64_encode(&raw);
+            assert_eq!(decode_i2p_b64(&text).unwrap(), raw, "{text}");
+            let plus = text.replace('-', "+");
+            let slash = text.replace('~', "/");
+            assert_eq!(decode_i2p_b64(&plus).unwrap(), raw, "{plus}");
+            assert_eq!(decode_i2p_b64(&slash).unwrap(), raw, "{slash}");
+        }
+    }
+
+    async fn sam_reply(lines: &'static [&'static str]) -> SocketAddr {
+        let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+        let addr = listener.local_addr().unwrap();
+        tokio::spawn(async move {
+            loop {
+                let Ok((mut sock, _)) = listener.accept().await else {
+                    break;
+                };
+                tokio::spawn(async move {
+                    for reply in lines {
+                        if read_line(&mut sock).await.is_err() {
+                            return;
+                        }
+                        if write_line(&mut sock, reply).await.is_err() {
+                            return;
+                        }
+                    }
+                });
+            }
+        });
+        addr
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn hello_requires_the_reply_prefix_and_result_ok() {
+        let ok = sam_reply(&["HELLO REPLY RESULT=OK VERSION=3.1"]).await;
+        let mut sock = TcpStream::connect(ok).await.unwrap();
+        hello(&mut sock).await.unwrap();
+
+        let missing = sam_reply(&["HELLO REPLY RESULT=NOVERSION"]).await;
+        let mut sock = TcpStream::connect(missing).await.unwrap();
+        let err = hello(&mut sock).await.unwrap_err();
+        assert!(err.to_string().contains("i2p sam hello"), "{err}");
+
+        let bare = sam_reply(&["RESULT=OK"]).await;
+        let mut sock = TcpStream::connect(bare).await.unwrap();
+        let err = hello(&mut sock).await.unwrap_err();
+        assert!(err.to_string().contains("i2p sam hello"), "{err}");
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn read_line_stops_at_16kib() {
+        async fn once(n: usize) -> Result<usize, NetError> {
+            use tokio::io::AsyncWriteExt;
+            let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let addr = listener.local_addr().unwrap();
+            let server = tokio::spawn(async move {
+                let (mut sock, _) = listener.accept().await.unwrap();
+                let body = vec![b'a'; n];
+                sock.write_all(&body).await.unwrap();
+                sock.write_all(b"\n").await.unwrap();
+            });
+            let mut sock = TcpStream::connect(addr).await.unwrap();
+            let got = read_line(&mut sock).await.map(|s| s.len());
+            let _ = server.await;
+            got
+        }
+        assert_eq!(once(16 * 1024).await.unwrap(), 16 * 1024);
+        assert!(once(16 * 1024 + 1).await.is_err());
+        assert_eq!(once(16 + 1024 + 1).await.unwrap(), 16 + 1024 + 1);
+    }
+
+    #[tokio::test(flavor = "current_thread")]
+    async fn stream_status_error_is_not_retried() {
+        let addr = sam_reply(&[
+            "HELLO REPLY RESULT=OK VERSION=3.1",
+            "STREAM STATUS RESULT=I2P_ERROR",
+        ])
+        .await;
+        let dialer = I2pDialer {
+            sam_addr: addr,
+            session_id: "once".into(),
+            destination: "dest".into(),
+            forward_port: None,
+        };
+        let started = std::time::Instant::now();
+        let err = dialer.stream_connect("peer.b32.i2p").await.unwrap_err();
+        assert!(
+            started.elapsed() < std::time::Duration::from_secs(2),
+            "a SAM status error is returned, not slept on"
+        );
+        assert!(err.to_string().contains("i2p sam stream"), "{err}");
+
+        let forward_addr = sam_reply(&[
+            "HELLO REPLY RESULT=OK VERSION=3.1",
+            "STREAM STATUS RESULT=I2P_ERROR",
+        ])
+        .await;
+        let (control, _held) = {
+            let hold = TcpListener::bind("127.0.0.1:0").await.unwrap();
+            let hold_addr = hold.local_addr().unwrap();
+            let accepted = tokio::spawn(async move {
+                let _ = hold.accept().await;
+            });
+            let control = TcpStream::connect(hold_addr).await.unwrap();
+            (control, accepted)
+        };
+        let mut sam = I2pSam {
+            sam_addr: forward_addr,
+            session_id: "fwd".into(),
+            destination: "dest".into(),
+            forward_port: None,
+            _control: control,
+            _forward: None,
+        };
+        let started = std::time::Instant::now();
+        let err = sam.stream_forward(9).await.unwrap_err();
+        assert!(
+            started.elapsed() < std::time::Duration::from_secs(2),
+            "a forward status error is returned, not slept on"
+        );
+        assert!(err.to_string().contains("i2p sam forward"), "{err}");
+    }
 }