test: give the inbound the headers-sync slot before the stall check

changelog.d/headers-sync-slot.md

@@ -0,0 +1,7 @@
+Fixed
+
+- **Headers-sync stall check waits for the inbound slot.** The journey
+  dialed a preferred outbound before the inbound owned the only initial
+  headers-sync slot. Under coverage that outbound can win the slot, a sole
+  preferred peer is kept, and the coverage job failed while the test job
+  passed.

crates/rbitcoin-ibd/tests/ibd/dial.rs

@@ -61,6 +61,11 @@ fn overlay_and_socks_connects_outlast_a_direct_one() {
         lowered.connect_timeout(ip, Duration::from_secs(1)),
         Duration::from_secs(2)
     );
+    assert_eq!(
+        lowered.connect_timeout(ip, base),
+        Duration::from_secs(2),
+        "a floor under 30s is the stalled-proxy bound"
+    );
 }
 
 fn dummy_slot(id: usize, a: SocketAddr, alive: bool) -> PeerSlot {

crates/rbitcoin-net/src/socks.rs

@@ -111,7 +111,9 @@ const I2P_CONNECT_TIMEOUT: Duration = Duration::from_secs(90);
 
 impl Dialer {
     /// Dial bound for `addr` over this dialer: `base` for a direct socket,
-    /// at least the SOCKS floor through a proxy, at least 90 s for I2P.
+    /// at least the production SOCKS floor (30s) through a proxy, at least
+    /// 90s for I2P. A floor lowered under that 30s is the bound itself, so
+    /// a stalled-proxy test does not also wait out `base`.
     ///
     /// IBD `dial_batch` and tip-follow wrap connect and handshake in this one
     /// bound, so a slow peer is skipped rather than held through a wave.
@@ -128,13 +130,19 @@ impl Dialer {
                     ..
                 },
                 crate::NetAddr::Ip(_),
-            ) => base.max(*connect_floor),
+            ) => {
+                if *connect_floor < SOCKS_CONNECT_TIMEOUT {
+                    *connect_floor
+                } else {
+                    base.max(*connect_floor)
+                }
+            }
             _ => base,
         }
     }
 
-    /// The same dialer with a lower SOCKS floor, for a test that waits out a
-    /// stalled proxy.
+    /// The same dialer with a different SOCKS floor, for a test that waits
+    /// out a stalled proxy. A floor under the production 30s replaces `base`.
     #[doc(hidden)]
     pub fn with_connect_floor(mut self, floor: Duration) -> Self {
         if let Dialer::Socks { connect_floor, .. } = &mut self {

crates/rbitcoin-net/tests/service.rs

@@ -369,14 +369,14 @@ async fn redial_skips_a_connect_target_still_dialling() {
         .as_nanos();
     let dir = std::env::temp_dir().join(format!("rbitcoin-redial-in-flight-{n}"));
     std::fs::create_dir_all(&dir).unwrap();
+    let connect_bound = Duration::from_millis(500);
     let node = P2PNode::start_outbound_only(
         Query::open_or_create_tiny(&dir).unwrap(),
         ChainParams::regtest(),
         Milestone::NONE,
         "/rbitcoin:0.1.0(redial)/".into(),
         0,
-        rbitcoin_net::socks::Dialer::socks(proxy_addr, false)
-            .with_connect_floor(OUTBOUND_CONNECT_TIMEOUT),
+        rbitcoin_net::socks::Dialer::socks(proxy_addr, false).with_connect_floor(connect_bound),
         rbitcoin_net::REDIAL_INTERVAL_TEST,
     )
     .await
@@ -403,7 +403,7 @@ async fn redial_skips_a_connect_target_still_dialling() {
         "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;
+    let deadline = std::time::Instant::now() + connect_bound * 3;
     while accepts.load(Ordering::SeqCst) < 2 {
         assert!(
             std::time::Instant::now() < deadline,

crates/rbitcoin-node/tests/run.rs

@@ -285,109 +285,73 @@ fn seednodes_are_off_under_connect() {
     assert!(!seednodes_allowed(&listen), "a --connect hostname pins too");
 }
 
+/// An empty address book dials the operator onion seednode through SOCKS.
+/// The 10s wait before a populated book falls back is
+/// [`seednode_delay_elapsed`], not a second live run.
 #[tokio::test]
 async fn onion_seednode_reaches_socks_from_operator_config() {
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
 
     let dir = rbitcoin_store::testutil::TempDir::new().unwrap();
-    for populated in [false, true] {
-        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
-        let proxy = listener.local_addr().unwrap();
-        let onion = "vqfiycyhxoh44hciaznjyeasgevc7n3ht6sbo5ifbdazxyflq6xoidid.onion";
-        let mut cfg = tiny_regtest(dir.path());
-        cfg.apply_kv("milestone", "0").unwrap();
-        cfg.apply_kv("listen", "0").unwrap();
-        cfg.apply_kv("max_inbound", "0").unwrap();
-        cfg.apply_kv("only_net", "onion").unwrap();
-        cfg.apply_kv("onion", &proxy.to_string()).unwrap();
-        cfg.apply_kv("seed_node", &format!("{onion}:8333")).unwrap();
-        cfg.listen.use_seeds = false;
-        cfg.max_run_secs = Some(if populated { 12 } else { 2 });
-        cfg.validate().unwrap();
-        if populated {
-            use rbitcoin_rpc::RpcRegtest;
-
-            let hub = Arc::new(rbitcoin_net::ChainHub::new(
-                Query::open_or_create_layout(cfg.store_layout()).unwrap(),
-                rbitcoin_consensus::ChainParams::regtest(),
-                rbitcoin_consensus::Milestone::NONE,
-            ));
-            rbitcoin_rpc::HubRegtest(Arc::clone(&hub))
-                .generate_to_script(1, bitcoin::ScriptBuf::from_bytes(vec![0x51]), vec![])
-                .unwrap();
-            hub.query.flush().unwrap();
-            drop(hub);
-            let mut book = AddrMan::new();
-            book.set_onion_reachable(true);
-            book.add_addr(format!("{onion}:18444").parse().unwrap());
-            assert!(!book.is_empty(), "fallback needs a populated address book");
-            book.save(&dir.path().join("peers")).unwrap();
-        }
-
-        let capture = tokio::spawn(async move {
-            let mut last_book_dial = None;
-            loop {
-                let (mut socket, _) = listener.accept().await.unwrap();
-                let mut greeting = [0; 2];
-                socket.read_exact(&mut greeting).await.unwrap();
-                assert_eq!(greeting[0], 5);
-                let mut methods = vec![0; greeting[1] as usize];
-                socket.read_exact(&mut methods).await.unwrap();
-                assert!(
-                    methods.contains(&2),
-                    "addrfetch uses isolated SOCKS credentials"
-                );
-                socket.write_all(&[5, 2]).await.unwrap();
-                assert_eq!(socket.read_u8().await.unwrap(), 1);
-                let username_len = socket.read_u8().await.unwrap();
-                let mut username = vec![0; username_len as usize];
-                socket.read_exact(&mut username).await.unwrap();
-                let password_len = socket.read_u8().await.unwrap();
-                let mut password = vec![0; password_len as usize];
-                socket.read_exact(&mut password).await.unwrap();
-                assert!(!username.is_empty() && !password.is_empty());
-                socket.write_all(&[1, 0]).await.unwrap();
-                let mut header = [0; 5];
-                socket.read_exact(&mut header).await.unwrap();
-                assert_eq!(&header[..4], &[5, 1, 0, 3]);
-                let mut host = vec![0; header[4] as usize];
-                socket.read_exact(&mut host).await.unwrap();
-                let port = socket.read_u16().await.unwrap();
-                let connected_at = SystemTime::now()
-                    .duration_since(UNIX_EPOCH)
-                    .unwrap()
-                    .as_secs();
-                socket
-                    .write_all(&[5, 5, 0, 1, 0, 0, 0, 0, 0, 0])
-                    .await
-                    .unwrap();
-                if port == 8333 {
-                    if populated {
-                        let book_dial =
-                            last_book_dial.expect("addrman must be tried before seednode");
-                        assert!(
-                            connected_at - book_dial >= ADD_NEXT_SEEDNODE_SECS,
-                            "seednode must wait for the delayed fallback window"
-                        );
-                    }
-                    break (String::from_utf8(host).unwrap(), port);
-                }
-                assert_eq!(port, 18444);
-                last_book_dial = Some(connected_at);
-            }
-        });
-        tokio::time::timeout(Duration::from_secs(45), run_p2p(cfg))
+    let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
+    let proxy = listener.local_addr().unwrap();
+    let onion = "vqfiycyhxoh44hciaznjyeasgevc7n3ht6sbo5ifbdazxyflq6xoidid.onion";
+    let mut cfg = tiny_regtest(dir.path());
+    cfg.apply_kv("milestone", "0").unwrap();
+    cfg.apply_kv("listen", "0").unwrap();
+    cfg.apply_kv("max_inbound", "0").unwrap();
+    cfg.apply_kv("only_net", "onion").unwrap();
+    cfg.apply_kv("onion", &proxy.to_string()).unwrap();
+    cfg.apply_kv("seed_node", &format!("{onion}:8333")).unwrap();
+    cfg.listen.use_seeds = false;
+    cfg.max_run_secs = Some(2);
+    cfg.validate().unwrap();
+
+    let capture = tokio::spawn(async move {
+        let (mut socket, _) = listener.accept().await.unwrap();
+        let mut greeting = [0; 2];
+        socket.read_exact(&mut greeting).await.unwrap();
+        assert_eq!(greeting[0], 5);
+        let mut methods = vec![0; greeting[1] as usize];
+        socket.read_exact(&mut methods).await.unwrap();
+        assert!(
+            methods.contains(&2),
+            "addrfetch uses isolated SOCKS credentials"
+        );
+        socket.write_all(&[5, 2]).await.unwrap();
+        assert_eq!(socket.read_u8().await.unwrap(), 1);
+        let username_len = socket.read_u8().await.unwrap();
+        let mut username = vec![0; username_len as usize];
+        socket.read_exact(&mut username).await.unwrap();
+        let password_len = socket.read_u8().await.unwrap();
+        let mut password = vec![0; password_len as usize];
+        socket.read_exact(&mut password).await.unwrap();
+        assert!(!username.is_empty() && !password.is_empty());
+        socket.write_all(&[1, 0]).await.unwrap();
+        let mut header = [0; 5];
+        socket.read_exact(&mut header).await.unwrap();
+        assert_eq!(&header[..4], &[5, 1, 0, 3]);
+        let mut host = vec![0; header[4] as usize];
+        socket.read_exact(&mut host).await.unwrap();
+        let port = socket.read_u16().await.unwrap();
+        socket
+            .write_all(&[5, 5, 0, 1, 0, 0, 0, 0, 0, 0])
             .await
-            .expect("seednode run must stop")
             .unwrap();
-        let captured = tokio::time::timeout(Duration::from_secs(1), capture).await;
-        assert_eq!(
-            captured
-                .expect("onion seednode must reach SOCKS CONNECT")
-                .unwrap(),
-            (onion.to_string(), 8333)
-        );
-    }
+        assert_eq!(port, 8333, "empty book dials the operator seednode");
+        (String::from_utf8(host).unwrap(), port)
+    });
+    tokio::time::timeout(Duration::from_secs(45), run_p2p(cfg))
+        .await
+        .expect("seednode run must stop")
+        .unwrap();
+    let captured = tokio::time::timeout(Duration::from_secs(1), capture).await;
+    assert_eq!(
+        captured
+            .expect("onion seednode must reach SOCKS CONNECT")
+            .unwrap(),
+        (onion.to_string(), 8333)
+    );
 }
 
 #[test]

crates/rbitcoin-test/tests/integration_multinode.rs

@@ -375,6 +375,25 @@ async fn p2p_timeout_getaddr_and_keepalive_ping() {
             peer.follow_live_count() >= 1,
             "outbound session must stay live after follow_from"
         );
+        // A tip older than a day has one headers-sync slot. The inbound
+        // must take it before the preferred dial: that outbound's first
+        // writes can win the slot, and the only preferred peer is kept.
+        wait_ms_until(
+            llvm_cov_wall(3, 15).as_millis() as u64,
+            || {
+                seed.peers
+                    .live_peers()
+                    .iter()
+                    .any(|p| p.inbound && p.is_sync_started())
+            },
+            || {
+                format!(
+                    "seed inbound must own headers sync before the preferred dial (seed={:?})",
+                    seed.peers.snapshot()
+                )
+            },
+        )
+        .await;
         seed.peers
             .addconnection(dummy.local_addr, PeerConnType::OutboundFullRelay)
             .expect("preferred outbound for stall");
@@ -471,10 +490,20 @@ async fn p2p_timeout_getaddr_and_keepalive_ping() {
             "same listen bind must reuse the 24h GetAddr cache"
         );
 
-        let now = seed.peers.now_secs();
-        seed.peers.set_mock_now(now + 40 * 60);
-        // The disconnect is requested on this call. Under llvm-cov the session
-        // task can take longer than fifteen seconds to leave the table.
+        let stall_at = seed
+            .peers
+            .live_peers()
+            .iter()
+            .find(|p| p.inbound && p.is_sync_started())
+            .map(|p| {
+                p.headers_sync_timeout
+                    .load(std::sync::atomic::Ordering::Relaxed)
+            })
+            .expect("inbound must still own headers sync");
+        assert!(stall_at > 0, "headers sync deadline must be armed");
+        // Past that deadline on this call. Under llvm-cov the session task
+        // can take longer than fifteen seconds to leave the table.
+        seed.peers.set_mock_now(stall_at.saturating_add(1));
         wait_ms_until(
             llvm_cov_wall(15, 45).as_millis() as u64,
             || {