diff --git a/TESTING.md b/TESTING.md index 9ea362761e..d90e6679db 100644 --- a/TESTING.md +++ b/TESTING.md @@ -390,7 +390,7 @@ A cell that names a crate-local test describes the witness that exists today. It | `a_heavier_header_chain_keeps_its_context` | IBD (crate) | One hub: a challenger keeps median time while the tip is genesis (invalid at the median, valid one second later); that hub then confirms a stale ancestor, a heavier chain from below it replaces the candidate and drops its script skip, and a rewind uses the confirmed base work and height. Retarget is a second hub (period length is not regtest difficulty): a heavier fork across a gap retarget is adopted, rewind restores difficulty before the next retarget, and a challenger keeps its retarget snapshot. Those three checkpoint layouts cannot be one candidate. | | `analog_selector_keeps_each_boundary_rate` | Mempool (crate) | One selector table: the exact band edge, absolute log distance, the requested quantile, and the two-hundredth nearest window each keep their own rate. | | `fee_history_gaps_cache_and_reorg_share_one_history` | IBD (crate) | One history: heights without a hurdle are not observations, rates stay cached until a new height, only the newest hashes are kept, and a reorg drops those above. The RPC backfill stays `fee_history_backfills_from_the_chain_when_relay_starts`. | -| `peer_header_dos_and_self_announce` | P2P (**default**, crate) | One hub and one peer: verack order (wtxidrelay and sendaddrv2 stick, ping is logged, redundant verack is ignored, sendaddrv2 and oversized addrv2 after verack disconnect), unknown-parent block and compact, a full header batch continues from its last header, minchainwork stays silent until the floor, bad proof-of-work disconnects and time-too-new does not, empty-locator serves only a hash that has a body, an ancient weaker header disconnects unless noban. A noban peer stays up on a bad block and gathers no score. A version-3 header at CLTV activation logs `bad-version`, and a far stamp logs `time-too-new`. | +| `peer_header_dos_and_self_announce` | P2P (**default**, crate) | One hub and one peer: verack order (wtxidrelay and sendaddrv2 stick, ping is logged, redundant verack is ignored, sendaddrv2 and oversized addrv2 after verack disconnect), unknown-parent block and compact, a full header batch continues from its last header, minchainwork stays silent until the floor, bad proof-of-work disconnects and time-too-new does not, empty-locator serves only a hash that has a body, an ancient weaker header disconnects unless noban. A noban or manual peer stays up on a bad block and gathers no score, and a manual peer that sends `sendaddrv2` after verack is still dropped. A version-3 header at CLTV activation logs `bad-version`, and a far stamp logs `time-too-new`. | | `self_announce_clearnet_then_overlay` | P2P (**default**, crate) | One address book, no chain: externalip is daily, loopback and `--no-discover` suppress it, then onion and i2p replace that clearnet address on the same book. | | `p2p_compact_hb_getblocktxn_and_orphan` | P2P (**default**) | One mature pad: HB coinbase `cmpctblock`, 2-tx compact → `getblocktxn` then same-peer compact retry while pending + `blocktxn` connect, unique short-id fill that fails header merkle → `getdata` (not `getblocktxn`, header not `BLOCK_FAILED`) then honest full `block` connects, orphan child GetData then parent accept (INV AlreadyHave), then live `getblocks` → `inv`, inbound `feefilter`, `filterload` disconnect. The pad seals filters through height 1, does not advertise `NODE_COMPACT_FILTERS` while the tip is ahead, stays silent for a stop at the tip, and answers `getcfilters` / `getcfheaders` / `getcfcheckpt` for the sealed height. An oversize locator is rejected. `mempool`, `filteradd`, and `filterclear` disconnect. Live mutated `block` disconnects in `on_block`. Inbound `getdata` of 20 witness blocks serves `MAX_SERVE_BLOCKS` (16); 17th is not queued (`getdata_skips_reconstruct_when_serve_inflight_at_cap`). Same-peer pending compact skip is `same_peer_pending_cmpct_does_not_getblocktxn_again`. Tokio-worker park and park-not-reject logs are `peer_blocksonly_and_orphan_tx` | | `p2p_feeler_completes_and_closes` | P2P (**default**) | Outbound feeler: VERSION then close (`feeler connection completed`). No live follow; dummy has no completed inbound. Same test: `run_feeler_timed` silence is `Timeout`. Inbound/outbound/plain silence stay `handshake_timeout_after_silence` | @@ -407,7 +407,7 @@ A cell that names a crate-local test describes the witness that exists today. It | `end_of_ibd_follow` | P2P (**default**) | Miner plus `--sh-index` syncer. One mature regtest: coinbases pay script A, one spend pays script B. IBD reaches that tip, leaves IBD, and Electrum history matches. The next block tip-follows. Restart does not rewrite the scripthash pack mark; one more block arrives by write-behind. Cancelling IBD once the height is below the miner, then `run_p2p`, finishes the same tip and history. With the syncer caught up, dropping the miner leaves tip follow (not IBD). A partial datadir whose miner is down does not open Electrum and stays short of that tip; the same miner address coming back lets that datadir finish | | `end_of_ibd_sh_interrupt` | P2P (**default**) | Same miner and scripts. The syncer first catches up with `--sh-index` off, then the datadir is frozen after pass 1 (`DONE.keys`, no `DONE.post`). Resume builds the index and Electrum's first answer is the full A/B history; restart does not rewrite the pack mark. A block mined after the freeze is in that history. A copy that already has the block, with `include_hwm` at the new tip and that create appended on the unsealed head, still serves the same history and does not open Electrum early | | `end_of_ibd_work_fork` | P2P (**default**) | Two miners on a regtest with retargeting and min-difficulty. One chain retargets harder and stops shorter. The other forks after that retarget, resets to the pow limit, and grows taller with less work. The syncer IBD-adopts the heavier tip, keeps it across a reopen, and follows one more heavy block while the tall chain is still ahead | -| `node_run_p2p_short` | Node (**default**) | Product `run_p2p` `--blocks-only` `--connect` to a live seeder (`--max-tip-age` so the 3-block pad is not stale IBD); process `getpeerinfo` / `getconnectioncount` / `getnetworkinfo` / `getnettotals` / `ping` while connected (v2 `manual`, as Core reports `-connect`; handshake `startingheight` equals the seeder tip; `timeoffset` present; `synced_headers`/`synced_blocks` stay `-1` until the peer announces a header hash (empty getheaders at tip does not copy VERSION height); `servicesnames` present; `getnetworkinfo.timeoffset` present); after catch-up `localrelay` / mempool `relay_enabled` stay false and `sendrawtransaction` is not `relay disabled`; Electrum `broadcast` and Esplora `POST /tx` admit decode/consensus errors (not hub-missing / not `relay disabled`); `addconnection inbound` refuses; `disconnectnode` unknown `nodeid` / empty params error then a real addr drops that row from the next `getpeerinfo`, the seeder sees it go, and a second `disconnectnode` of that `nodeid` (or of an address never connected) is `-29`; the `--connect` peer is redialled as `manual` under a new id and the seeder sees the inbound; seeder inbound `tx` then ends that session. Exit via `stop`. `max_run_secs=0` is `node_listen_and_exit`. Mock-clock `timeoffset` is the median (odd N, even N upper-middle, inbound-only 0, peer clock behind). A connecting dummy reports `-1`. Header-only and connected peers report `synced_blocks` differently. A query without a chain errors. `pingwait`, `NETWORK_LIMITED`, and `noban` are on the peer row | +| `node_run_p2p_short` | Node (**default**) | Product `run_p2p` `--blocks-only` `--connect` to a live seeder (`--max-tip-age` so the 3-block pad is not stale IBD); process `getpeerinfo` / `getconnectioncount` / `getnetworkinfo` / `getnettotals` / `ping` while connected (v2 `manual`, as Core reports `-connect`; handshake `startingheight` equals the seeder tip; `timeoffset` present; `synced_headers`/`synced_blocks` stay `-1` until the peer announces a header hash (empty getheaders at tip does not copy VERSION height); `servicesnames` present; `getnetworkinfo.timeoffset` present); after catch-up `localrelay` / mempool `relay_enabled` stay false and `sendrawtransaction` is not `relay disabled`; Electrum `broadcast` and Esplora `POST /tx` admit decode/consensus errors (not hub-missing / not `relay disabled`); `addconnection inbound` refuses; `disconnectnode` unknown `nodeid` / empty params error then a real addr drops that row from the next `getpeerinfo`, the seeder sees it go, and a second `disconnectnode` of that `nodeid` (or of an address never connected) is `-29`; the `--connect` peer is redialled as `manual` under a new id and the seeder sees the inbound; `addnode onetry` and `add` of that address succeed and open no second session; seeder inbound `tx` then ends that session. Exit via `stop`. `max_run_secs=0` is `node_listen_and_exit`. Mock-clock `timeoffset` is the median (odd N, even N upper-middle, inbound-only 0, peer clock behind). A connecting dummy reports `-1`. Header-only and connected peers report `synced_blocks` differently. A query without a chain errors. `pingwait`, `NETWORK_LIMITED`, and `noban` are on the peer row | Removed (covered by the rows above): `confirm_cross_block_prevout_without_tx_head`, `double_archive_keeps_tx_height_for_coinbase_maturity`, `mega_batch_duplicate_header_is_idempotent`, diff --git a/changelog.d/manual-peer-parity.md b/changelog.d/manual-peer-parity.md new file mode 100644 index 0000000000..0fc7f364ae --- /dev/null +++ b/changelog.d/manual-peer-parity.md @@ -0,0 +1,15 @@ +Fixed + +- **Manual peers are not punished for misbehavior, as in Core.** A + `--connect` or `addnode` peer that sends an invalid or mutated block, an + oversized `inv`, `getdata` or `addrv2`, or a bad compact block now stays + connected and its address is not refused. Core never disconnects or + discourages a manual peer for misbehavior. Protocol violations that Core + answers with a plain disconnect, such as `sendaddrv2` after `verack` or a + `tx` to a `--blocks-only` node, still drop a manual peer. +- **`addnode` opens no second session to a connected address.** `addnode + onetry` or `add` of an address with a live session, or one still being + dialled, dialled it again, and each dial that connected became its own + session. As in Core, the RPC now succeeds without dialling. A dial now + counts from the moment it is queued, so an `addnode` or `--connect` + redial right behind another dial to the same address adds nothing. diff --git a/crates/rbitcoin-net/src/peer.rs b/crates/rbitcoin-net/src/peer.rs index 6e8379836c..f4e4cc2ea3 100644 --- a/crates/rbitcoin-net/src/peer.rs +++ b/crates/rbitcoin-net/src/peer.rs @@ -140,11 +140,11 @@ async fn accept_received_from_peer( }) } -/// One-day in-memory refusal. Noban peers are not recorded. +/// One-day in-memory refusal. Noban and manual peers are not recorded. /// A loopback address is every local connection, so Core disconnects that /// peer and does not discourage the address. fn note_threshold_refusal(session: Option<&crate::peers::LivePeer>) { - let Some(s) = session.filter(|s| !s.session_noban()) else { + let Some(s) = session.filter(|s| !misbehavior_exempt(s)) else { return; }; if local_addr_skips_discourage(s.addr.ip()) { @@ -169,6 +169,27 @@ fn threshold_disconnect(session: Option<&crate::peers::LivePeer>) -> NetError { NetError::Protocol("peer misbehavior threshold") } +/// Core `MaybeDiscourageAndDisconnect`: misbehavior never disconnects or +/// discourages a noban peer or a manual one (`--connect`, `addnode`). +fn misbehavior_exempt(s: &crate::peers::LivePeer) -> bool { + s.session_noban() || s.conn_type == crate::peers::PeerConnType::Manual +} + +/// Core `Misbehaving`. Unlike [`punish_disconnect`], it also keeps a manual +/// peer connected. +fn misbehaving(ban_score: &mut u32, session: Option<&crate::peers::LivePeer>) { + if let Some(s) = + session.filter(|s| s.conn_type == crate::peers::PeerConnType::Manual && !s.session_noban()) + { + rbitcoin_log::info!("Warning: not punishing manually connected peer {}!", s.id); + return; + } + punish_disconnect(ban_score, session); +} + +/// Disconnects and refuses the address; a noban peer is kept. Used where Core +/// sets `fDisconnect`, which drops a manual peer too. Core `Misbehaving` +/// sites call [`misbehaving`]. fn punish_disconnect(ban_score: &mut u32, session: Option<&crate::peers::LivePeer>) { if let Some(s) = session.filter(|s| s.session_noban()) { rbitcoin_log::info!("Warning: not punishing noban peer {}!", s.id); @@ -181,9 +202,10 @@ fn punish_disconnect(ban_score: &mut u32, session: Option<&crate::peers::LivePee } } -/// Core `MaybeDiscourageAndDisconnect`: misbehavior never drops a noban peer. +/// Core `MaybeDiscourageAndDisconnect`: misbehavior never drops a noban or +/// manual peer. fn misbehavior_disconnects(ban_score: u32, session: Option<&crate::peers::LivePeer>) -> bool { - ban_score >= BAN_SCORE_THRESHOLD && !session.is_some_and(|s| s.session_noban()) + ban_score >= BAN_SCORE_THRESHOLD && !session.is_some_and(misbehavior_exempt) } /// Cap on incomplete compact blocks awaiting `blocktxn` (DoS). const MAX_PENDING_CMPCT: usize = 1; @@ -2889,7 +2911,7 @@ fn on_addrv2( rbitcoin_log::info!("{}", received_addrv2_log(nbytes, id)); if n > MAX_ADDR_TO_SEND { rbitcoin_log::info!("{}", addrv2_message_size_log(n)); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } if session.is_some_and(|s| s.conn_type == crate::peers::PeerConnType::AddrFetch && n > 1) { @@ -3026,7 +3048,7 @@ async fn serve_getdata( inv: &[Inventory], ) -> Result<(), NetError> { if inv.len() > MAX_INV_SIZE { - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } let inflight = session.map(|s| &s.serve_inflight); @@ -3202,11 +3224,8 @@ fn on_getblocktxn( } if bad { rbitcoin_log::info!("p2p: getblocktxn with out-of-bounds tx indices"); - // Out-of-range indexes: disconnect. - follow.ban_score = follow.ban_score.saturating_add(BAN_SCORE_THRESHOLD); - if let Some(s) = session { - s.request_disconnect(); - } + // Core `Misbehaving`: noban and manual peers stay connected. + misbehaving(&mut follow.ban_score, session); } else { // Core: past `MAX_GETBLOCKTXN_DEPTH` (10) send the full block. const MAX_GETBLOCKTXN_DEPTH: u32 = 10; @@ -3244,7 +3263,7 @@ fn on_inv( items: &[Inventory], ) -> Result<(), NetError> { if items.len() > MAX_INV_SIZE { - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } let mut want = Vec::new(); @@ -3434,7 +3453,7 @@ fn on_headers( { // Headers on a cached-invalid chain: disconnect // (`p2p_unrequested_blocks` step 8 follow-up header). - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } let connecting = header_announcement_connects(hub, &follow.pending_headers, prev); @@ -3520,19 +3539,19 @@ async fn on_block( if !block.check_merkle_root() { rbitcoin_log::info!("Block mutated: bad-txnmrklroot, hashMerkleRoot mismatch"); take_requested_block(hub, &mut follow.requested_blocks, &hash); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } if crate::compact::merkle_body_mutated(&block.txdata) { rbitcoin_log::info!("Block mutated: bad-txns-duplicate"); take_requested_block(hub, &mut follow.requested_blocks, &hash); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } if rbitcoin_consensus::block_mutated_without_coinbase(block) { rbitcoin_log::info!("Block mutated: 64-byte transaction without a coinbase"); take_requested_block(hub, &mut follow.requested_blocks, &hash); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } if let Some(s) = session { @@ -3561,7 +3580,7 @@ async fn on_block( } if let Err(e) = hub.ensure_header(&block.header) { if !crate::chain::accept_err_is_temporary_time(&e) { - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); } return Ok(()); } @@ -3592,12 +3611,12 @@ fn on_block_unrequested_skip( && !follow.pending_headers.contains_key(&prev) { rbitcoin_log::info!("{}", accept_prev_not_found_log(hash)); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(true); } if let Err(e) = hub.ensure_header(&block.header) { if !crate::chain::accept_err_is_temporary_time(&e) { - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); } return Ok(true); } @@ -3747,7 +3766,7 @@ fn on_cmpctblock_reject_early( ) -> Result { if session.is_some_and(|s| s.has_failed_cmpct(&hash)) { rbitcoin_log::info!("p2p: previous compact block reconstruction attempt failed"); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(true); } if let Some(s) = session { @@ -3757,13 +3776,13 @@ fn on_cmpctblock_reject_early( } if !crate::compact::prefilled_indexes_ok(hsi) { rbitcoin_log::info!("p2p: invalid index in cmpctblock message"); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(true); } // Child of a cached-invalid block: disconnect. Same-hash cached invalid // via compact stays connected. if hub.is_block_invalid(&hsi.header.prev_blockhash) { - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(true); } if hub.is_block_invalid(&hash) { @@ -3956,7 +3975,7 @@ async fn on_blocktxn( } if session.is_some_and(|s| s.has_failed_cmpct(&hash)) { rbitcoin_log::info!("p2p: previous compact block reconstruction attempt failed"); - punish_disconnect(&mut follow.ban_score, session); + misbehaving(&mut follow.ban_score, session); return Ok(()); } let Some(pc) = follow.pending_cmpct.remove(&hash) else { diff --git a/crates/rbitcoin-net/src/peer_header_dos_journey.rs b/crates/rbitcoin-net/src/peer_header_dos_journey.rs index ed6a80886e..be06a0ba87 100644 --- a/crates/rbitcoin-net/src/peer_header_dos_journey.rs +++ b/crates/rbitcoin-net/src/peer_header_dos_journey.rs @@ -801,8 +801,9 @@ fn empty_locator_needs_a_body(hub: &crate::chain::ChainHub) { ); } -/// A bad block disconnects a plain peer. noban keeps the session and scores nothing. -async fn noban_bad_block_is_not_punished( +/// A bad block disconnects a plain peer. noban and manual peers keep the +/// session and score nothing (Core `MaybeDiscourageAndDisconnect`). +async fn noban_and_manual_bad_block_is_not_punished( hub: &crate::chain::ChainHub, peers: &std::sync::Arc, ) { @@ -861,6 +862,30 @@ async fn noban_bad_block_is_not_punished( assert!(misbehavior_disconnects(BAN_SCORE_THRESHOLD, None)); assert!(!misbehavior_disconnects(BAN_SCORE_THRESHOLD - 1, None)); peers.set_noban(false); + + // `--connect` and `addnode` peers are `manual`. + let manual = live_peer_as(peers, 18477, 26, crate::peers::PeerConnType::Manual); + let mut follow = PeerFollowState::new(); + follow.requested_blocks.insert(bad_pow.block_hash()); + on_block(hub, &out_tx, &mut follow, Some(manual.as_ref()), &bad_pow) + .await + .unwrap(); + assert!( + !manual.stop.load(Ordering::SeqCst), + "Core never disconnects a manual peer for misbehavior" + ); + assert_eq!(follow.ban_score, 0, "a manual peer gathers no score"); + assert!( + !misbehavior_disconnects(BAN_SCORE_THRESHOLD, Some(manual.as_ref())), + "a manual peer at the threshold stays connected" + ); + // Core answers `sendaddrv2` after `verack` with `fDisconnect`, not + // `Misbehaving`, so a manual peer is still dropped. + on_sendaddrv2(&mut follow, Some(manual.as_ref())); + assert!( + manual.stop.load(Ordering::SeqCst), + "a protocol violation still drops a manual peer" + ); } /// Core `IsBlockMutated`: a coinbase-less body with a 64-byte tx is dropped @@ -1275,7 +1300,7 @@ async fn peer_header_dos_and_self_announce() { tall_unknown_parent_skips_reconstruct(&hub, &peers).await; header_reject_punishes_except_time(&hub).await; - noban_bad_block_is_not_punished(&hub, &peers).await; + noban_and_manual_bad_block_is_not_punished(&hub, &peers).await; sixty_four_byte_body_is_mutated(&hub, &peers).await; merkle_mismatch_forgets_the_ask(&hub, &peers).await; duplicate_pair_disconnects(&hub, &peers).await; diff --git a/crates/rbitcoin-net/src/peer_tests.rs b/crates/rbitcoin-net/src/peer_tests.rs index a33ab88dd9..c24fefee9a 100644 --- a/crates/rbitcoin-net/src/peer_tests.rs +++ b/crates/rbitcoin-net/src/peer_tests.rs @@ -3117,6 +3117,28 @@ fn threshold_exit_refuses_the_address_without_punish_disconnect() { !peers.inbound_discouraged(std::net::SocketAddr::from(([1, 2, 3, 4], 1))), "noban is not recorded" ); + peers.set_noban(false); + + // Core never discourages a manual peer. + let manual_addr = std::net::SocketAddr::from(([5, 6, 7, 8], 8333)); + let manual = peers.register( + manual_addr, + manual_addr, + &ver, + false, + crate::peers::PeerConnType::Manual, + ); + let _ = threshold_disconnect(Some(manual.as_ref())); + let mut score = 0u32; + punish_disconnect(&mut score, Some(manual.as_ref())); + assert!( + manual.stop.load(std::sync::atomic::Ordering::SeqCst), + "a protocol violation drops a manual peer" + ); + assert!( + !peers.inbound_discouraged(std::net::SocketAddr::from(([5, 6, 7, 8], 1))), + "a manual peer is not recorded" + ); } #[test] diff --git a/crates/rbitcoin-net/src/peer_tip_announce_journey.rs b/crates/rbitcoin-net/src/peer_tip_announce_journey.rs index 7ce4765115..0fd5beda65 100644 --- a/crates/rbitcoin-net/src/peer_tip_announce_journey.rs +++ b/crates/rbitcoin-net/src/peer_tip_announce_journey.rs @@ -90,6 +90,20 @@ fn live_peer( port: u16, nonce: u64, inbound: bool, +) -> std::sync::Arc { + let conn = if inbound { + crate::peers::PeerConnType::Inbound + } else { + crate::peers::PeerConnType::OutboundFullRelay + }; + live_peer_as(peers, port, nonce, conn) +} + +fn live_peer_as( + peers: &std::sync::Arc, + port: u16, + nonce: u64, + conn: crate::peers::PeerConnType, ) -> std::sync::Arc { use bitcoin::p2p::address::Address; use bitcoin::p2p::message_network::VersionMessage; @@ -105,11 +119,7 @@ fn live_peer( start_height: 0, relay: true, }; - let conn = if inbound { - crate::peers::PeerConnType::Inbound - } else { - crate::peers::PeerConnType::OutboundFullRelay - }; + let inbound = conn == crate::peers::PeerConnType::Inbound; peers.register(addr, addr, &ver, inbound, conn) } diff --git a/crates/rbitcoin-net/src/peers.rs b/crates/rbitcoin-net/src/peers.rs index c7f074d48b..1f660a5fcd 100644 --- a/crates/rbitcoin-net/src/peers.rs +++ b/crates/rbitcoin-net/src/peers.rs @@ -204,18 +204,15 @@ impl std::fmt::Display for DialTarget { /// An outbound dial counted in [`PeerHub::dial_in_flight`]; dropping it /// ends the count. +#[derive(Debug)] pub(crate) struct DialInFlight { - peers: Arc, + counts: Arc>>, key: String, } impl Drop for DialInFlight { fn drop(&mut self) { - let mut g = self - .peers - .dials_in_flight - .lock() - .unwrap_or_else(|e| e.into_inner()); + let mut g = self.counts.lock().unwrap_or_else(|e| e.into_inner()); if let Some(n) = g.get_mut(&self.key) { *n -= 1; if *n == 0 { @@ -226,10 +223,14 @@ impl Drop for DialInFlight { } /// Request that the node dial `addr` as `typ`. -#[derive(Clone, Debug)] +#[derive(Debug)] pub struct DialRequest { pub target: DialTarget, pub typ: PeerConnType, + /// Counts the dial from the moment it is queued, so a redial or + /// `addnode` that runs before the dial task picks it up sees it. `None` + /// for a feeler, which registers no session and is never counted. + pub(crate) in_flight: Option, } #[derive(Clone, Debug)] @@ -1296,7 +1297,7 @@ pub struct PeerHub { connect_addrs: Mutex>, /// Outbound dials still before session registration, by target. The /// redial pass skips these, so a slow connect is not stacked every 2 s. - dials_in_flight: Mutex>, + dials_in_flight: Arc>>, /// Network P2P port used when a remembered host omits `:port`. `0` = unset. connect_default_port: AtomicU16, dial_tx: Mutex>>, @@ -1413,7 +1414,7 @@ impl PeerHub { manual_hosts: Mutex::new(HashSet::new()), connect_hosts: Mutex::new(Vec::new()), connect_addrs: Mutex::new(Vec::new()), - dials_in_flight: Mutex::new(HashMap::new()), + dials_in_flight: Arc::new(Mutex::new(HashMap::new())), connect_default_port: AtomicU16::new(0), dial_tx: Mutex::new(None), hb_selected: Mutex::new(Vec::new()), @@ -2358,10 +2359,22 @@ impl PeerHub { } fn dial_manual_net(&self, addr: &crate::NetAddr) -> Result<(), String> { - match addr.socket_addr() { - Some(ip) => self.dial(ip, PeerConnType::Manual), - None => self.dial_domain(addr.host_str(), addr.port(), PeerConnType::Manual), + self.dial_manual_target(DialTarget::from_net(*addr)) + } + + /// Core `AlreadyConnectedToHost` / `AlreadyConnectedToAddressPort`: + /// `addnode` and the redial pass open no second session to an endpoint + /// that is connected or still dialling, and the RPC still succeeds. + fn dial_manual_target(&self, target: DialTarget) -> Result<(), String> { + // Claim before the live check: a dial registers its session before + // it drops its claim, so no window shows the endpoint as neither. + let Some(claim) = self.claim_dial(&target) else { + return Ok(()); + }; + if self.is_target_live(&target) { + return Ok(()); } + self.send_dial(target, PeerConnType::Manual, Some(claim)) } pub fn addnode_net(&self, addr: crate::NetAddr, cmd: &str) -> Result<(), String> { @@ -2394,13 +2407,13 @@ impl PeerHub { pub fn addnode_host(&self, node: &str, cmd: &str, default_port: u16) -> Result<(), String> { self.note_default_port(default_port); match cmd { - "onetry" => self.dial_resolved(node, PeerConnType::Manual), + "onetry" => self.dial_resolved(node), "add" => { self.manual_hosts .lock() .unwrap_or_else(|e| e.into_inner()) .insert(node.to_string()); - let _ = self.dial_resolved(node, PeerConnType::Manual); + let _ = self.dial_resolved(node); Ok(()) } "remove" => { @@ -2451,7 +2464,7 @@ impl PeerHub { Ok(DialTarget::Socket(addr)) } - fn dial_resolved(&self, node: &str, typ: PeerConnType) -> Result<(), String> { + fn dial_resolved(&self, node: &str) -> Result<(), String> { let target = self.host_dial_target(node)?; match &target { DialTarget::Socket(addr) => { @@ -2467,14 +2480,7 @@ impl PeerHub { .insert(target.net_addr()); } } - self.dial_target(target, typ) - } - - fn dial_target(&self, target: DialTarget, typ: PeerConnType) -> Result<(), String> { - match target { - DialTarget::Socket(addr) => self.dial(addr, typ), - DialTarget::Domain { host, port } => self.dial_domain(host, port, typ), - } + self.dial_manual_target(target) } fn disconnect_target(&self, target: &DialTarget) -> bool { @@ -2497,10 +2503,11 @@ impl PeerHub { } } - /// Count an outbound dial to `target` until the guard drops. Held from - /// before the TCP / SOCKS connect until the session is registered or the - /// dial fails, whichever path ends it. - pub(crate) fn dial_in_flight(self: &Arc, target: &DialTarget) -> DialInFlight { + /// Count an outbound dial to `target` until the guard drops. Taken when + /// the dial is queued (a direct follow dial takes it before the TCP / + /// SOCKS connect) and held until the session is registered or the dial + /// fails, whichever path ends it. + pub(crate) fn dial_in_flight(&self, target: &DialTarget) -> DialInFlight { let key = target.to_string(); *self .dials_in_flight @@ -2509,16 +2516,27 @@ impl PeerHub { .entry(key.clone()) .or_insert(0) += 1; DialInFlight { - peers: Arc::clone(self), + counts: Arc::clone(&self.dials_in_flight), key, } } - fn is_target_dialing(&self, target: &DialTarget) -> bool { - self.dials_in_flight + /// [`Self::dial_in_flight`], or `None` when `target` already has a dial + /// queued or connecting. Check and count are one step. + fn claim_dial(&self, target: &DialTarget) -> Option { + let key = target.to_string(); + let mut g = self + .dials_in_flight .lock() - .unwrap_or_else(|e| e.into_inner()) - .contains_key(&target.to_string()) + .unwrap_or_else(|e| e.into_inner()); + if g.contains_key(&key) { + return None; + } + g.insert(key.clone(), 1); + Some(DialInFlight { + counts: Arc::clone(&self.dials_in_flight), + key, + }) } fn is_target_live(&self, target: &DialTarget) -> bool { @@ -2571,10 +2589,7 @@ impl PeerHub { if !seen.insert(target.to_string()) { continue; } - if self.is_target_live(&target) || self.is_target_dialing(&target) { - continue; - } - let _ = self.dial_target(target, PeerConnType::Manual); + let _ = self.dial_manual_target(target); } } @@ -2663,11 +2678,31 @@ impl PeerHub { } pub fn dial_net(&self, addr: crate::NetAddr, typ: PeerConnType) -> Result<(), String> { + self.queue_dial(DialTarget::from_net(addr), typ) + } + + /// Queue a dial, counted in flight from now on. A feeler is not counted: + /// it registers no session, and its probe has no connect deadline, so a + /// stuck one would block the redial pass for that endpoint. + fn queue_dial(&self, target: DialTarget, typ: PeerConnType) -> Result<(), String> { + let in_flight = (typ != PeerConnType::Feeler).then(|| self.dial_in_flight(&target)); + self.send_dial(target, typ, in_flight) + } + + /// Queue a dial. The request carries `in_flight` to the dial task; if it + /// is never sent, dropping it ends the count. + fn send_dial( + &self, + target: DialTarget, + typ: PeerConnType, + in_flight: Option, + ) -> Result<(), String> { let g = self.dial_tx.lock().unwrap_or_else(|e| e.into_inner()); let tx = g.as_ref().ok_or("no dialer attached")?; tx.send(DialRequest { - target: DialTarget::from_net(addr), + target, typ, + in_flight, }) .map_err(|_| "dialer closed".to_string()) } @@ -2795,13 +2830,7 @@ impl PeerHub { } pub fn dial(&self, addr: SocketAddr, typ: PeerConnType) -> Result<(), String> { - let g = self.dial_tx.lock().unwrap_or_else(|e| e.into_inner()); - let tx = g.as_ref().ok_or("no dialer attached")?; - tx.send(DialRequest { - target: DialTarget::from_net(crate::NetAddr::Ip(addr)), - typ, - }) - .map_err(|_| "dialer closed".to_string()) + self.dial_net(crate::NetAddr::Ip(addr), typ) } pub fn dial_domain( @@ -2810,16 +2839,11 @@ impl PeerHub { port: u16, typ: PeerConnType, ) -> Result<(), String> { - let g = self.dial_tx.lock().unwrap_or_else(|e| e.into_inner()); - let tx = g.as_ref().ok_or("no dialer attached")?; - tx.send(DialRequest { - target: DialTarget::Domain { - host: host.into(), - port, - }, - typ, - }) - .map_err(|_| "dialer closed".to_string()) + let target = DialTarget::Domain { + host: host.into(), + port, + }; + self.queue_dial(target, typ) } } @@ -3525,6 +3549,8 @@ mod tests { let got = take_dials(&mut rx); assert_eq!(got.len(), 1, "{got:?}"); assert!(matches!(got[0].typ, PeerConnType::Manual), "{got:?}"); + // A queued request counts as in flight; dropping it ends that dial. + drop(got); // A --connect address (IP or overlay) redials in the same pass. One // that meets a hostname at an endpoint shares its dial, and a live @@ -3541,6 +3567,7 @@ mod tests { assert!(got.iter().all(|d| d.typ == PeerConnType::Manual), "{got:?}"); assert_eq!(got[0].target, DialTarget::Socket(ip), "{got:?}"); assert_eq!(got[1].target, DialTarget::from_net(onion), "{got:?}"); + drop(got); let _live = hub.register( ip, ip, @@ -3554,6 +3581,84 @@ mod tests { assert_eq!(got[0].target, DialTarget::from_net(onion), "{got:?}"); } + #[test] + fn addnode_skips_an_endpoint_that_is_connected_or_dialling() { + let hub = PeerHub::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + hub.set_dialer(tx); + let ip = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 18446); + let live = hub.register( + ip, + ip, + &ver("/rbitcoin:0.1.0/"), + false, + PeerConnType::OutboundFullRelay, + ); + // Core `AlreadyConnectedToHost` / `AlreadyConnectedToAddressPort`. + hub.addnode_host("127.0.0.1:18446", "onetry", 18444) + .unwrap(); + hub.addnode_host("127.0.0.1:18446", "add", 18444).unwrap(); + hub.addnode(ip, "onetry").unwrap(); + let got = take_dials(&mut rx); + assert!( + got.is_empty(), + "a connected endpoint is not dialled again: {got:?}" + ); + + let onion: crate::NetAddr = + "pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion:8333" + .parse() + .unwrap(); + let dialing = hub.dial_in_flight(&DialTarget::from_net(onion)); + hub.addnode_host(&onion.to_string(), "onetry", 18444) + .unwrap(); + let got = take_dials(&mut rx); + assert!(got.is_empty(), "nor is one still dialling: {got:?}"); + drop(dialing); + // A dial counts from the moment it is queued, before the dial task + // picks it up: `onetry`, `add` and a redial pass right behind it + // queue nothing more. + hub.addnode_host(&onion.to_string(), "onetry", 18444) + .unwrap(); + hub.addnode_host(&onion.to_string(), "add", 18444).unwrap(); + hub.redial_remembered(); + let got = take_dials(&mut rx); + assert_eq!(got.len(), 1, "a queued dial is not queued again: {got:?}"); + assert_eq!(got[0].target, DialTarget::from_net(onion), "{got:?}"); + + live.request_disconnect(); + hub.addnode_host("127.0.0.1:18446", "onetry", 18444) + .unwrap(); + let got = take_dials(&mut rx); + assert_eq!(got.len(), 1, "once the session ends, onetry dials: {got:?}"); + assert_eq!(got[0].target, DialTarget::Socket(ip), "{got:?}"); + assert_eq!(got[0].typ, PeerConnType::Manual, "{got:?}"); + } + + #[test] + fn a_queued_feeler_does_not_hold_off_a_manual_dial() { + let hub = PeerHub::new(); + let (tx, mut rx) = mpsc::unbounded_channel(); + hub.set_dialer(tx); + let ip = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 18447); + // A feeler's probe has no connect deadline. Counted in flight, a + // stuck one would keep `addnode` and the redial pass off this + // endpoint until it ended. + hub.addconnection(ip, PeerConnType::Feeler).unwrap(); + let feeler = take_dials(&mut rx); + assert_eq!(feeler.len(), 1, "{feeler:?}"); + assert_eq!(feeler[0].typ, PeerConnType::Feeler, "{feeler:?}"); + hub.addnode(ip, "onetry").unwrap(); + let got = take_dials(&mut rx); + assert_eq!( + got.len(), + 1, + "a queued feeler does not block addnode: {got:?}" + ); + assert_eq!(got[0].typ, PeerConnType::Manual, "{got:?}"); + drop(feeler); + } + #[tokio::test] async fn slow_redial_resolve_does_not_stall_runtime() { let hub = PeerHub::new(); diff --git a/crates/rbitcoin-net/src/service.rs b/crates/rbitcoin-net/src/service.rs index 2b48f4f59d..acee60ef33 100644 --- a/crates/rbitcoin-net/src/service.rs +++ b/crates/rbitcoin-net/src/service.rs @@ -206,7 +206,17 @@ impl P2PNode { let (ah_tx, ah_rx) = tokio::sync::oneshot::channel::(); let h = tokio::spawn(async move { let _ = run_outbound_session_with_abort( - req.target, magic, local_addr, hub, peers, ua, live, req.typ, ah_rx, d, + req.target, + req.in_flight, + magic, + local_addr, + hub, + peers, + ua, + live, + req.typ, + ah_rx, + d, ) .await; }); @@ -365,6 +375,7 @@ impl P2PNode { let target = DialTarget::from_net(peer); let prepared = prepare_outbound_session( target, + None, self.magic, self.local_addr, self.hub.clone(), @@ -610,6 +621,7 @@ const OUTBOUND_CONNECT_TIMEOUT: Duration = Duration::from_secs(2); #[allow(clippy::too_many_arguments)] // call-site args stay unbundled async fn prepare_outbound_session( peer: DialTarget, + in_flight: Option, magic: Magic, local: SocketAddr, hub: Arc, @@ -620,8 +632,9 @@ async fn prepare_outbound_session( dialer: crate::socks::Dialer, ) -> Result { rbitcoin_log::debug!("{}", crate::peers::trying_connection_log(typ, &peer)); - // Until this returns, the redial pass treats `peer` as already dialling. - let _dialing = peers.dial_in_flight(&peer); + // Until this returns, the redial pass and `addnode` treat `peer` as + // already dialling. A queued dial was counted when it was queued. + let _dialing = in_flight.unwrap_or_else(|| peers.dial_in_flight(&peer)); let peer_net = peer.net_addr(); let stream = tokio::time::timeout( crate::connect_timeout_for(peer_net, OUTBOUND_CONNECT_TIMEOUT), @@ -726,6 +739,7 @@ async fn run_prepared_outbound(prepared: PreparedOutbound) -> Result<(), NetErro #[allow(clippy::too_many_arguments)] // call-site args stay unbundled async fn run_outbound_session_with_abort( peer: DialTarget, + in_flight: Option, magic: Magic, local: SocketAddr, hub: Arc, @@ -749,6 +763,7 @@ async fn run_outbound_session_with_abort( } let prepared = prepare_outbound_session( peer, + in_flight, magic, local, hub, diff --git a/crates/rbitcoin-test/tests/integration_multinode.rs b/crates/rbitcoin-test/tests/integration_multinode.rs index bec307523e..033b763f92 100644 --- a/crates/rbitcoin-test/tests/integration_multinode.rs +++ b/crates/rbitcoin-test/tests/integration_multinode.rs @@ -3002,6 +3002,22 @@ async fn node_run_p2p_short() { ) .await; + // Core `AlreadyConnectedToHost`: `addnode` to a connected + // endpoint succeeds and opens no second session. + for cmd in ["onetry", "add"] { + let r = jsonrpc(rpc_addr, "addnode", json!([addr.clone(), cmd])).await; + assert!(r["error"].is_null(), "addnode {cmd}: {r}"); + } + let quiet = Instant::now() + Duration::from_millis(500); + while Instant::now() < quiet { + let peers = jsonrpc(rpc_addr, "getpeerinfo", json!([])).await; + let n = peers["result"].as_array().map_or(0, |rows| { + rows.iter().filter(|p| p["addr"] == addr.as_str()).count() + }); + assert_eq!(n, 1, "addnode opened a second session to {addr}: {peers}"); + tokio::time::sleep(Duration::from_millis(50)).await; + } + pin_blocksonly_seeder_tx_disconnects(rpc_addr, &seed_peers).await; let _ = jsonrpc(rpc_addr, "stop", json!([])).await; });