net: redial --connect addresses that drop, as in Core

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`,

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.

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<PeerHub>,
+    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<HashSet<String>>,
     /// `--connect` hostnames that are not a [`crate::NetAddr`] (clearnet DNS).
     connect_hosts: Mutex<Vec<String>>,
+    /// `--connect` targets given as a [`crate::NetAddr`] (IP, onion, I2P, CJDNS).
+    connect_addrs: Mutex<Vec<crate::NetAddr>>,
+    /// 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<HashMap<String, usize>>,
     /// Network P2P port used when a remembered host omits `:port`. `0` = unset.
     connect_default_port: AtomicU16,
     dial_tx: Mutex<Option<mpsc::UnboundedSender<DialRequest>>>,
@@ -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<crate::NetAddr>) {
+        *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<Self>, 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<DialTarget, String>) {
         let mut seen = HashSet::<String>::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]

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<PreparedOutbound, NetError> {
     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);
+    }
 }

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(