Run the TCP blackhole tests on FreeBSD

The rx-stall tests and the background connect timeout test need a local
TCP address whose SYNs go unanswered. They build it from a listener whose
accept queue is full, which Linux, macOS and Windows leave unanswered.
FreeBSD answers such a SYN from its syncache and then resets the
connection, so the helper panicked and eleven tests failed in the FreeBSD
package build on every branch.

On FreeBSD the helper now fills by replacing the listener with a bound
socket that does not listen, and drains by listening on it. That address
is silent only with net.inet.tcp.blackhole=2 and blackhole_local=1, which
also stop every refused connect on the host, so the FreeBSD package
workflow runs the tests in two passes: everything else with the settings
off, then the blackhole tests alone with them on. A connect that is not
silent still fails the helper, naming the settings.

Once the helper worked, three tests failed on FreeBSD because they took
one read() as one frame: FreeBSD delivered the link announce that follows
a msg2 in a read of its own, where Linux returned the two together. The
tests now read each frame by its length with the transport's own frame
reader, and drain the announces after the handshake.

.github/workflows/package-freebsd.yml

@@ -99,7 +99,7 @@ jobs:
             cargo build --release
 
             # The only place the FreeBSD cfg arms' unit tests ever run in
-            # CI — the main CI matrix is Linux-only, and a release build
+            # CI — the main CI matrix runs no FreeBSD job, and a release build
             # compiles no #[cfg(test)] code (AF-prefix strip round-trips,
             # platform module, config path gates).
             #
@@ -116,7 +116,18 @@ jobs:
             # This bounds the damage; it does not fix anything. A test that can
             # block for ever is a defect at the test, and the ones this suite
             # has are bounded where they are written.
-            if ! timeout -s KILL 900 cargo test; then
+            #
+            # The tests that need a TCP address whose SYNs go unanswered
+            # (`testutil::Blackhole`) run in a second pass. FreeBSD resets a
+            # connect to a full accept queue, so it gets that address only
+            # from the kernel's blackhole settings, and those also stop every
+            # refused connect on the host, which other tests need. So the
+            # first pass skips them and the second runs only them, with the
+            # settings on.
+            BLACKHOLE_TESTS="node::tests::rx_stall:: transport::tcp::tests::background_connect_timeout_is_counted"
+            skips=""
+            for t in $BLACKHOLE_TESTS; do skips="$skips --skip $t"; done
+            if ! timeout -s KILL 900 cargo test -- $skips; then
               echo "FAIL: cargo test failed, or did not finish within its 900s bound." >&2
               echo "      If the output above stops mid-run, look for libtest's" >&2
               echo "      'has been running for over' lines: they name the test" >&2
@@ -124,6 +135,16 @@ jobs:
               echo "      are the rest of the answer." >&2
               exit 1
             fi
+            bh_was=$(sysctl -n net.inet.tcp.blackhole)
+            bh_local_was=$(sysctl -n net.inet.tcp.blackhole_local)
+            sysctl net.inet.tcp.blackhole=2 net.inet.tcp.blackhole_local=1
+            bh_rc=0
+            timeout -s KILL 300 cargo test --lib -- $BLACKHOLE_TESTS || bh_rc=$?
+            sysctl net.inet.tcp.blackhole="$bh_was" net.inet.tcp.blackhole_local="$bh_local_was"
+            if [ "$bh_rc" -ne 0 ]; then
+              echo "FAIL: the blackhole pass failed, or did not finish within its 300s bound." >&2
+              exit 1
+            fi
 
             packaging/freebsd/build-pkg.sh \
               --version "$FREEBSD_PACKAGE_VERSION" \

src/node/tests/rx_stall.rs

@@ -17,12 +17,13 @@
 //! it does not start toward an inbound peer's address, and that a later send
 //! uses the connection once it is up.
 //!
-//! The unanswered SYN is constructed locally: a listener with a backlog of
-//! zero whose single accept slot is already taken. Linux drops further SYNs to
-//! a listener whose accept queue is full, so a connect to it times out rather
-//! than being refused. `Blackhole::silent()` checks that before any test
-//! relies on it, which is what lets a regression show up at its real size:
-//! one connect timeout per reply, counted in `connect_timeouts`.
+//! The unanswered SYN is constructed locally by `Blackhole`: a listener whose
+//! accept queue is full, which Linux, macOS and Windows leave unanswered, or
+//! on FreeBSD a bound port with no listener and the kernel's blackhole
+//! settings on (see `Blackhole`). A connect to it times out rather than being
+//! refused. `Blackhole` checks that before any test relies on it, which is
+//! what lets a regression show up at its real size: one connect timeout per
+//! reply, counted in `connect_timeouts`.
 //!
 //! The tests print their measurements; run with `--nocapture` to see them.
 
@@ -155,24 +156,65 @@ async fn prime_link(node: &Node, bh: &Blackhole) -> std::net::TcpStream {
     accepted
 }
 
-/// Close the node's connection to `bh` from the far end, after taking the
-/// listener's accept slot so that any later dial to the address hangs.
+/// Close the node's connection to `bh` from the far end, after filling `bh`
+/// so that any later dial to the address hangs.
 async fn kill_link(node: &Node, bh: &mut Blackhole, accepted: std::net::TcpStream) {
     bh.fill();
     drop(accepted);
     wait_pool_gone(node, &bh.transport_addr()).await;
 }
 
-/// Read one frame's worth of bytes the node wrote to `far_end`, if any
-/// arrive within its read timeout.
-fn read_frame(far_end: &mut std::net::TcpStream) -> Option<Vec<u8>> {
-    let mut buf = [0u8; 2048];
-    match far_end.read(&mut buf) {
-        Ok(n) if n > 0 => Some(buf[..n].to_vec()),
-        _ => None,
+/// A blocking `std` stream read through tokio's `AsyncRead`, so the
+/// transport's own frame reader can read it from a synchronous helper. Its
+/// reads never return `Pending`.
+struct BlockingRead<'a>(&'a mut std::net::TcpStream);
+
+impl tokio::io::AsyncRead for BlockingRead<'_> {
+    fn poll_read(
+        self: std::pin::Pin<&mut Self>,
+        _cx: &mut std::task::Context<'_>,
+        buf: &mut tokio::io::ReadBuf<'_>,
+    ) -> std::task::Poll<std::io::Result<()>> {
+        let this = self.get_mut();
+        let n = this.0.read(buf.initialize_unfilled())?;
+        buf.advance(n);
+        std::task::Poll::Ready(Ok(()))
     }
 }
 
+/// Read one FMP frame the node wrote to `far_end`, if one arrives within its
+/// read timeout. The frame is read by its own length, not by what one `read`
+/// returns: TCP may hand over two frames in one read or one across two, and
+/// which it does differs by kernel (FreeBSD delivered the announce that
+/// follows a msg2 in a read of its own, where Linux returned the two together).
+fn read_frame(far_end: &mut std::net::TcpStream) -> Option<Vec<u8>> {
+    use std::future::Future;
+    let mut reader = BlockingRead(far_end);
+    let mut read = std::pin::pin!(crate::transport::framing::read_fmp_packet(
+        &mut reader,
+        u16::MAX
+    ));
+    let mut cx = std::task::Context::from_waker(std::task::Waker::noop());
+    match read.as_mut().poll(&mut cx) {
+        std::task::Poll::Ready(frame) => frame.ok(),
+        std::task::Poll::Pending => unreachable!("a blocking read never returns Pending"),
+    }
+}
+
+/// Read and discard every frame the node writes to `far_end` until none
+/// arrives for a while, so a later read sees only what a later send wrote.
+/// Promotion follows the msg2 with link announces, which would otherwise be
+/// read in place of the frame a test is waiting for.
+fn drain_frames(far_end: &mut std::net::TcpStream) {
+    far_end
+        .set_read_timeout(Some(Duration::from_millis(200)))
+        .unwrap();
+    while read_frame(far_end).is_some() {}
+    far_end
+        .set_read_timeout(Some(Duration::from_millis(1000)))
+        .unwrap();
+}
+
 /// What one handler call against a TCP reply address did.
 #[derive(Debug)]
 struct Reply {
@@ -346,9 +388,11 @@ async fn peer_on_tcp(bh: &Blackhole) -> (Node, Identity, NodeAddr, std::net::Tcp
     assert_eq!(p.transport_id(), Some(TransportId::new(TCP_ID)));
     assert_eq!(p.current_addr(), Some(&link));
     assert!(
-        read_frame(&mut far_end).is_some(),
+        read_frame(&mut far_end)
+            .is_some_and(|f| CommonPrefix::parse(&f).is_some_and(|p| p.phase == PHASE_MSG2)),
         "the msg2 went out on the connection"
     );
+    drain_frames(&mut far_end);
     (node, sender, sender_addr, far_end)
 }
 
@@ -540,8 +584,8 @@ async fn msg1_resend_to_dead_outbound_leg_recovers_after_background_connect() {
         "no background connect toward the dial address"
     );
 
-    // The address starts answering: empty the accept queue, and the
-    // background connect's retransmitted SYN completes.
+    // The address starts answering (`Blackhole::drain`), and the background
+    // connect's retransmitted SYN completes.
     let _filler_ends = bh.drain();
     let start = Instant::now();
     let mut tick = 1;

src/testutil.rs

@@ -99,12 +99,22 @@ pub(crate) async fn wait_until<F: FnMut() -> bool>(mut f: F, limit: std::time::D
 /// A local TCP address whose SYNs go unanswered once filled.
 ///
 /// A listener with a backlog of one whose accept queue is filled: Linux,
-/// macOS and the BSDs drop further SYNs to a listener whose accept queue is
+/// macOS and Windows drop further SYNs to a listener whose accept queue is
 /// full, so a connect to it times out rather than completing or being
 /// refused. How many connects the queue takes before it is full differs by
 /// kernel (one on Linux, more on macOS), so filling stops at the first
 /// connect that times out. The listener and the fillers must be kept alive
 /// for as long as that is relied on.
+///
+/// FreeBSD answers a SYN to a full queue from its syncache and then resets
+/// the connection, so its queue never goes silent. There, filling replaces
+/// the listener with a socket bound to the same address that does not
+/// listen, and the kernel must be set not to reset a SYN to a port nobody
+/// listens on: `net.inet.tcp.blackhole=2` and `net.inet.tcp.blackhole_local=1`
+/// (root). Those settings also stop every refused connect on the host, so a
+/// test that needs one cannot run in the same pass; the FreeBSD package
+/// workflow runs the tests that use a filled `Blackhole` in a pass of their
+/// own.
 pub(crate) struct Blackhole {
     pub(crate) listener: socket2::Socket,
     fillers: Vec<std::net::TcpStream>,
@@ -143,6 +153,7 @@ impl Blackhole {
     /// Fill the listener's accept queue, stopping at the first connect that
     /// times out, which shows a further connect now times out instead of
     /// completing or being refused.
+    #[cfg(not(target_os = "freebsd"))]
     pub(crate) fn fill(&mut self) {
         const MAX_FILLERS: usize = 64;
         for _ in 0..=MAX_FILLERS {
@@ -163,6 +174,7 @@ impl Blackhole {
     /// answers again: the next connect to it, or the next retransmitted SYN
     /// of one already waiting, completes. Returns the accepted far ends,
     /// which the caller keeps alive while it relies on that.
+    #[cfg(not(target_os = "freebsd"))]
     pub(crate) fn drain(&mut self) -> Vec<std::net::TcpStream> {
         self.fillers
             .iter()
@@ -170,6 +182,37 @@ impl Blackhole {
             .collect()
     }
 
+    /// Close the listener and bind a socket that does not listen to the same
+    /// address, then check that a connect to it now times out. Connections
+    /// the listener already accepted are left as they are.
+    #[cfg(target_os = "freebsd")]
+    pub(crate) fn fill(&mut self) {
+        use socket2::{Domain, Socket, Type};
+        let quiet = Socket::new(Domain::IPV4, Type::STREAM, None).unwrap();
+        quiet.set_reuse_address(true).unwrap();
+        drop(std::mem::replace(&mut self.listener, quiet));
+        self.listener.bind(&self.addr.into()).unwrap();
+        let probe =
+            std::net::TcpStream::connect_timeout(&self.addr, std::time::Duration::from_millis(200));
+        match probe {
+            Err(e) if e.kind() == std::io::ErrorKind::TimedOut => {}
+            other => panic!(
+                "blackhole is not silent: probe connect returned {other:?}; on FreeBSD \
+                 this needs net.inet.tcp.blackhole=2 and net.inet.tcp.blackhole_local=1, \
+                 and a test using it belongs in package-freebsd.yml's blackhole pass"
+            ),
+        }
+    }
+
+    /// Start listening on the bound socket, so the address answers again: the
+    /// next connect to it, or the next retransmitted SYN of one already
+    /// waiting, completes. Nothing was queued, so there are no far ends.
+    #[cfg(target_os = "freebsd")]
+    pub(crate) fn drain(&mut self) -> Vec<std::net::TcpStream> {
+        self.listener.listen(1).unwrap();
+        Vec::new()
+    }
+
     /// The address in the transport form.
     pub(crate) fn transport_addr(&self) -> crate::transport::TransportAddr {
         crate::transport::TransportAddr::from_string(&self.addr.to_string())