diff --git a/TESTING.md b/TESTING.md index 34e1fc2a4..0b80e63bd 100644 --- a/TESTING.md +++ b/TESTING.md @@ -375,7 +375,7 @@ Prefer **one high-level scenario** per behavior cluster. Delete lower-level test | `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` is `-29`; `addnode onetry` reconnects as `manual` and the seeder sees the inbound; seeder inbound `tx` then disconnects. Exit via `stop`. `max_run_secs=0` is `node_listen_and_exit`. Mock-clock `timeoffset` median (odd N, even N upper-middle, inbound-only 0, peer clock behind), connecting dummy `-1`, header-only vs connected `synced_blocks`, query-without-chain, `pingwait` / `NETWORK_LIMITED` / `noban` stay RPC guts | +| `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` median (odd N, even N upper-middle, inbound-only 0, peer clock behind), connecting dummy `-1`, header-only vs connected `synced_blocks`, query-without-chain, `pingwait` / `NETWORK_LIMITED` / `noban` stay RPC guts | 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/connect-redial.md b/changelog.d/connect-redial.md new file mode 100644 index 000000000..6ce2fe5a8 --- /dev/null +++ b/changelog.d/connect-redial.md @@ -0,0 +1,16 @@ +Fixed + +- **A `--connect` address is redialled when its session drops.** Only + `--connect` hostnames were retried. An IP, onion, I2P or CJDNS target got + one dial at startup, so a node whose `--connect` peer restarted or was + disconnected stayed without peers until it was restarted. As in Core, + once the node follows the tip, every `--connect` target without a live + session is redialled every 2 s, including any that the startup dial + skipped (it dials at most three). +- **A redial does not stack while a dial is still connecting.** The 2 s + redial of `--connect` and `addnode add` targets skips a target whose + previous dial has not finished connecting. Before, a target behind a slow + SOCKS circuit or a host that drops SYNs got a new dial every pass, and + each one that connected became its own session. An outbound connect now + gives up after 8 s (90 s for I2P), as IBD and tip-follow dials already + did, so a proxy that never answers cannot hold a target forever. diff --git a/crates/rbitcoin-net/src/peers.rs b/crates/rbitcoin-net/src/peers.rs index 97a23fcc3..c7f074d48 100644 --- a/crates/rbitcoin-net/src/peers.rs +++ b/crates/rbitcoin-net/src/peers.rs @@ -202,6 +202,29 @@ impl std::fmt::Display for DialTarget { } } +/// An outbound dial counted in [`PeerHub::dial_in_flight`]; dropping it +/// ends the count. +pub(crate) struct DialInFlight { + peers: 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()); + if let Some(n) = g.get_mut(&self.key) { + *n -= 1; + if *n == 0 { + g.remove(&self.key); + } + } + } +} + /// Request that the node dial `addr` as `typ`. #[derive(Clone, Debug)] pub struct DialRequest { @@ -1269,6 +1292,11 @@ pub struct PeerHub { manual_hosts: Mutex>, /// `--connect` hostnames that are not a [`crate::NetAddr`] (clearnet DNS). connect_hosts: Mutex>, + /// `--connect` targets given as a [`crate::NetAddr`] (IP, onion, I2P, CJDNS). + 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>, /// Network P2P port used when a remembered host omits `:port`. `0` = unset. connect_default_port: AtomicU16, dial_tx: Mutex>>, @@ -1384,6 +1412,8 @@ impl PeerHub { added: Mutex::new(HashSet::new()), 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()), connect_default_port: AtomicU16::new(0), dial_tx: Mutex::new(None), hb_selected: Mutex::new(Vec::new()), @@ -2392,6 +2422,13 @@ impl PeerHub { *self.connect_hosts.lock().unwrap_or_else(|e| e.into_inner()) = hosts; } + /// `--connect` targets that parse as a [`crate::NetAddr`]. The redial pass + /// dials each one that has no live session, as Core keeps reconnecting + /// every `-connect` address. + pub fn set_connect_addrs(&self, addrs: Vec) { + *self.connect_addrs.lock().unwrap_or_else(|e| e.into_inner()) = addrs; + } + fn note_default_port(&self, default_port: u16) { if default_port != 0 && self.connect_default_port.load(Ordering::Relaxed) == 0 { self.connect_default_port @@ -2460,6 +2497,30 @@ 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 { + let key = target.to_string(); + *self + .dials_in_flight + .lock() + .unwrap_or_else(|e| e.into_inner()) + .entry(key.clone()) + .or_insert(0) += 1; + DialInFlight { + peers: Arc::clone(self), + key, + } + } + + fn is_target_dialing(&self, target: &DialTarget) -> bool { + self.dials_in_flight + .lock() + .unwrap_or_else(|e| e.into_inner()) + .contains_key(&target.to_string()) + } + fn is_target_live(&self, target: &DialTarget) -> bool { let peers = self.snapshot(); match target { @@ -2490,19 +2551,27 @@ impl PeerHub { /// One `manual` dial per resolved endpoint in this pass, as Core dials /// `-addnode` and `-connect`. `addnode add` and `--connect` of the same - /// host share that dial. A later pass dials again when the session is + /// host share that dial, as do a `--connect` hostname and address that + /// meet at one endpoint. A later pass dials again when the session is /// still not live. DNS lookup is synchronous; callers on a Tokio worker /// use [`Self::redial_remembered_off_runtime`]. pub fn redial_remembered_with(&self, resolve: impl Fn(&str) -> Result) { let mut seen = HashSet::::new(); - for host in self.remembered_redials() { - let Ok(target) = resolve(&host) else { - continue; - }; + let addrs = self + .connect_addrs + .lock() + .unwrap_or_else(|e| e.into_inner()) + .clone(); + let targets = self + .remembered_redials() + .into_iter() + .filter_map(|host| resolve(&host).ok()) + .chain(addrs.into_iter().map(DialTarget::from_net)); + for target in targets { if !seen.insert(target.to_string()) { continue; } - if self.is_target_live(&target) { + if self.is_target_live(&target) || self.is_target_dialing(&target) { continue; } let _ = self.dial_target(target, PeerConnType::Manual); @@ -3456,6 +3525,33 @@ mod tests { let got = take_dials(&mut rx); assert_eq!(got.len(), 1, "{got:?}"); assert!(matches!(got[0].typ, PeerConnType::Manual), "{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 + // session is not dialled again. + let ip = SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 18445); + let onion: crate::NetAddr = + "pg6mmjiyjmcrsslvykfwnntlaru7p5svn6y2ymmju6nubxndf4pscryd.onion:8333" + .parse() + .unwrap(); + hub.set_connect_addrs(vec![crate::NetAddr::Ip(ip), onion]); + hub.redial_remembered(); + let got = take_dials(&mut rx); + assert_eq!(got.len(), 2, "{got:?}"); + 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:?}"); + let _live = hub.register( + ip, + ip, + &ver("/rbitcoin:0.1.0/"), + false, + PeerConnType::Manual, + ); + hub.redial_remembered(); + let got = take_dials(&mut rx); + assert_eq!(got.len(), 1, "only the onion has no live session: {got:?}"); + assert_eq!(got[0].target, DialTarget::from_net(onion), "{got:?}"); } #[tokio::test] diff --git a/crates/rbitcoin-net/src/service.rs b/crates/rbitcoin-net/src/service.rs index 2ee90df79..c0a644074 100644 --- a/crates/rbitcoin-net/src/service.rs +++ b/crates/rbitcoin-net/src/service.rs @@ -596,6 +596,16 @@ struct PreparedOutbound { id: u64, } +/// Bound on the TCP / SOCKS / SAM connect of an outbound dial, the same 8 s +/// as the IBD and tip-follow dials ([`crate::connect_timeout_for`] gives I2P +/// longer). Without it a proxy that accepts and never answers holds the +/// dial, and its [`crate::peers::DialInFlight`] mark, for good, so the +/// redial pass would never try that target again. +#[cfg(not(test))] +const OUTBOUND_CONNECT_TIMEOUT: Duration = Duration::from_secs(8); +#[cfg(test)] +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, @@ -609,8 +619,15 @@ 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); let peer_net = peer.net_addr(); - let stream = dialer.connect_net(peer_net).await?; + let stream = tokio::time::timeout( + crate::connect_timeout_for(peer_net, OUTBOUND_CONNECT_TIMEOUT), + dialer.connect_net(peer_net), + ) + .await + .map_err(|_| NetError::Timeout)??; let peer_hint = peer.version_socket(); let bind = stream.local_addr().unwrap_or(local); let height = hub.tip_height().map(|h| h as i32).unwrap_or(0); @@ -1019,4 +1036,74 @@ mod tests { "loopback client must show as inbound in getpeerinfo" ); } + + /// The 2 s redial pass must not stack a dial that is still connecting, + /// and must not lose the target to one that never finishes. A SOCKS + /// proxy that accepts and never answers holds the first dial in + /// `connect_net`: later passes skip it until the connect times out, then + /// dial again. + #[tokio::test(flavor = "multi_thread", worker_threads = 2)] + async fn redial_skips_a_connect_target_still_dialling() { + let _live = live_p2p_lock().await; + let proxy = TcpListener::bind("127.0.0.1:0").await.unwrap(); + let proxy_addr = proxy.local_addr().unwrap(); + let accepts = Arc::new(AtomicUsize::new(0)); + let seen = Arc::clone(&accepts); + let stall = tokio::spawn(async move { + let mut held = Vec::new(); + while let Ok((sock, _)) = proxy.accept().await { + seen.fetch_add(1, Ordering::SeqCst); + held.push(sock); + } + }); + let n = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap() + .as_nanos(); + let dir = std::env::temp_dir().join(format!("rbitcoin-redial-in-flight-{n}")); + std::fs::create_dir_all(&dir).unwrap(); + let node = P2PNode::start_outbound_only( + Query::open_or_create_tiny(&dir).unwrap(), + ChainParams::regtest(), + Milestone::NONE, + "/rbitcoin:0.1.0(redial)/".into(), + 0, + crate::socks::Dialer::socks(proxy_addr, false), + ) + .await + .unwrap(); + node.peers + .set_connect_addrs(vec![crate::NetAddr::Ip("127.0.0.1:18444".parse().unwrap())]); + node.peers.redial_remembered(); + let deadline = std::time::Instant::now() + Duration::from_secs(5); + while accepts.load(Ordering::SeqCst) == 0 { + assert!( + std::time::Instant::now() < deadline, + "the first dial never reached the proxy" + ); + tokio::time::sleep(Duration::from_millis(10)).await; + } + for _ in 0..3 { + node.peers.redial_remembered(); + } + tokio::time::sleep(Duration::from_millis(300)).await; + assert_eq!( + accepts.load(Ordering::SeqCst), + 1, + "a dial still connecting must not be dialled again" + ); + // The stalled connect times out and frees the target for the next pass. + let deadline = std::time::Instant::now() + OUTBOUND_CONNECT_TIMEOUT * 3; + while accepts.load(Ordering::SeqCst) < 2 { + assert!( + std::time::Instant::now() < deadline, + "a connect that never finishes must time out and be redialled" + ); + node.peers.redial_remembered(); + tokio::time::sleep(Duration::from_millis(100)).await; + } + stall.abort(); + drop(node); + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/crates/rbitcoin-node/src/run.rs b/crates/rbitcoin-node/src/run.rs index 0d79f8992..99a158938 100644 --- a/crates/rbitcoin-node/src/run.rs +++ b/crates/rbitcoin-node/src/run.rs @@ -788,6 +788,10 @@ pub async fn run_p2p(config: NodeConfig) -> Result<(), NodeError> { warn!("node: no follow peers connected — tip announce may stall"); } } + // From here the 2 s redial pass reconnects any `--connect` address + // without a live session, including ones past `follow_n`. Hostnames + // were registered before catch-up (`set_connect_hosts`). + node.peers.set_connect_addrs(config.listen.connect.clone()); } let (mut electrum_handles, mut electrum_bridge, electrum_onion) = start_electrum_if_ready( diff --git a/crates/rbitcoin-test/tests/integration_multinode.rs b/crates/rbitcoin-test/tests/integration_multinode.rs index 49ed161be..2abb8cbc3 100644 --- a/crates/rbitcoin-test/tests/integration_multinode.rs +++ b/crates/rbitcoin-test/tests/integration_multinode.rs @@ -2774,6 +2774,16 @@ async fn pin_blocksonly_seeder_tx_disconnects( script_pubkey: ScriptBuf::from_bytes(vec![0x51]), }], }; + // The node redials its `--connect` peer, so wait for these sessions to + // end rather than for getpeerinfo to be empty. + let before = jsonrpc(rpc_addr, "getpeerinfo", json!([])).await; + let ids: Vec = before["result"] + .as_array() + .expect("getpeerinfo array") + .iter() + .filter_map(|p| p["id"].as_u64()) + .collect(); + assert!(!ids.is_empty(), "{before}"); wait_ms_until( 3_000, || { @@ -2794,7 +2804,10 @@ async fn pin_blocksonly_seeder_tx_disconnects( let gone_deadline = Instant::now() + Duration::from_secs(5); loop { let peers = jsonrpc(rpc_addr, "getpeerinfo", json!([])).await; - if peers["result"].as_array().is_some_and(|a| a.is_empty()) { + if peers["result"].as_array().is_some_and(|a| { + a.iter() + .all(|p| p["id"].as_u64().is_none_or(|id| !ids.contains(&id))) + }) { break; } if Instant::now() >= gone_deadline { @@ -2940,6 +2953,7 @@ async fn node_run_p2p_short() { ); let id = rows[0]["id"].as_u64().expect("id"); + let bind = rows[0]["addrbind"].as_str().expect("addrbind").to_string(); let disc = jsonrpc(rpc_addr, "disconnectnode", json!([addr.clone()])).await; assert!(disc["error"].is_null(), "{disc}"); let after = jsonrpc(rpc_addr, "getpeerinfo", json!([])).await; @@ -2949,55 +2963,48 @@ async fn node_run_p2p_short() { .is_some_and(|a| a.iter().all(|p| p["id"].as_u64() != Some(id))), "disconnectnode must clear the row before the session task exits: {after}" ); - let again = jsonrpc(rpc_addr, "disconnectnode", json!([addr.clone()])).await; + // That session id is gone for good; the redial gets a new one. + let again = jsonrpc(rpc_addr, "disconnectnode", json!({"nodeid": id})).await; assert_eq!(again["error"]["code"], -29, "node not connected: {again}"); - let gone_deadline = Instant::now() + Duration::from_secs(5); - loop { - let peers = jsonrpc(rpc_addr, "getpeerinfo", json!([])).await; - if peers["result"].as_array().is_some_and(|a| a.is_empty()) { - break; - } - if Instant::now() >= gone_deadline { - panic!("getpeerinfo still occupied after disconnectnode: {peers}"); - } - tokio::time::sleep(Duration::from_millis(20)).await; - } + let stranger = jsonrpc(rpc_addr, "disconnectnode", json!(["127.0.0.1:1"])).await; + assert_eq!( + stranger["error"]["code"], -29, + "address not connected: {stranger}" + ); wait_ms_until( 3_000, || { !seed_peers .snapshot() .into_iter() - .any(|p| p.inbound && p.handshake_complete) + .any(|p| p.inbound && p.addr.to_string() == bind) }, || { format!( - "the seeder must see the disconnect: {:?}", + "the seeder must see the disconnect of {bind}: {:?}", seed_peers.snapshot() ) }, ) .await; - let added = jsonrpc( - rpc_addr, - "addnode", - json!([seed_addr.to_string(), "onetry"]), - ) - .await; - assert!(added["error"].is_null(), "{added}"); + // Core keeps reconnecting a `-connect` address that drops. let re_deadline = Instant::now() + Duration::from_secs(8); loop { let peers = jsonrpc(rpc_addr, "getpeerinfo", json!([])).await; - let ok = peers["result"].as_array().is_some_and(|rows| { - rows.iter() - .any(|p| p["inbound"] == false && p["connection_type"] == "manual") + let back = peers["result"].as_array().is_some_and(|rows| { + rows.iter().any(|p| { + p["addr"] == addr.as_str() + && p["inbound"] == false + && p["connection_type"] == "manual" + && p["id"].as_u64() != Some(id) + }) }); - if ok { + if back { break; } if Instant::now() >= re_deadline { - panic!("addnode onetry did not produce a manual peer: {peers}"); + panic!("--connect peer was not redialled after disconnectnode: {peers}"); } tokio::time::sleep(Duration::from_millis(50)).await; } @@ -3011,7 +3018,7 @@ async fn node_run_p2p_short() { }, || { format!( - "the seeder must see the addnode peer inbound: {:?}", + "the seeder must see the redialled peer inbound: {:?}", seed_peers.snapshot() ) },