fix(transport/udp): let connected sockets join an adopted traversal socket's port

A NAT traversal binds its base socket with a plain port-zero bind, and a
successful punch adopts that socket as the peer's transport. Connected-UDP
activation then opens a per-peer connected socket on the transport's local
address. Linux admits that second bind only when the socket already holding
the port set a reuse flag, and the traversal socket set neither, so every
activation attempt failed with EADDRINUSE at bind. The rx-loop tick retried
it on every pass, and the peer never left the unconnected path.

Set SO_REUSEPORT and SO_REUSEADDR inside UdpRawSocket::adopt, after the
socket is already bound. The order matters. Flags set before a port-zero
bind let the kernel hand out a port another flagged socket already holds, so
two concurrent traversals could share a port and the kernel would silently
split one peer's datagrams across two transports. Flags set after the bind
cannot change which port the socket was given; they only let a later socket
join it. The listen socket in UdpRawSocket::open already sets its flags
after its bind for the same reason.

adopt has one caller chain, which ends at traversal adoption, so the STUN
probe and configured listeners are untouched. The flags are best-effort, as
in open: if setting them fails, adoption still succeeds and only the
connected fast path is refused, rather than a working punched peer being
torn down.

Two tests cover the change, and both fail without it. The first adopts a
socket bound the way the traversal path binds, checks that it arrives with
neither flag, and requires a connected socket to open on its port and both
flags to read back; without the change the open fails with EADDRINUSE at
bind. The second adopts a traversal socket into a node peered with a node on
a configured listener, runs activation on both, and requires a connected
socket on each side, the adopted side's on the adopted socket's own port;
without the change the configured side activates, the adopted side does
not, and the failure carries the bind error from a direct open.

Each part of the change was also removed in turn. Without SO_REUSEPORT the
first test fails on that readback while the connected open still succeeds,
because either flag on the holder admits the join; without SO_REUSEADDR it
fails on that readback the same way; without both, both tests fail at the
bind. Flagging the socket before adoption fails the first test's
precondition, and running the second with FIPS_CONNECTED_UDP=0 fails its
configured-listener assertion first, so neither can pass vacuously.

Not established: Darwin behaviour. adopt is shared by the Unix backends, so
macOS builds get the flags too, and no measurement covers SO_REUSEPORT set
after bind there. FIPS_MACOS_CONNECTED_UDP=0 or FIPS_CONNECTED_UDP=0 turns
the fast path off without a rebuild.

CHANGELOG.md

@@ -17,6 +17,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   decrypt-worker completion path already acted on that return; the in-line
   decrypt path discarded it, so the socket stayed installed and the send path
   kept preferring it over the wildcard listen socket.
+- A peer reached by NAT traversal now gets its per-peer connected UDP socket.
+  The adopted traversal socket carried no address-reuse flags, so the connected
+  socket's bind to the same port was refused with `EADDRINUSE` on every tick and
+  the peer never left the unconnected path. The flags are now set when the
+  socket is adopted, after its bind, so the traversal bind still receives a
+  port no other socket holds.
 
 #### Peering
 

src/node/tests/bootstrap.rs

@@ -377,3 +377,119 @@ async fn test_adopted_udp_inherits_mtu_from_named_primary_config() {
         transport.stop().await.ok();
     }
 }
+
+/// A peer reached through an adopted traversal socket gets a per-peer
+/// connected UDP socket on that socket's own port, as a peer on a configured
+/// listener does. The traversal socket comes from a plain bind with no reuse
+/// flags, and the kernel refuses the connected socket's bind to a port whose
+/// holder did not opt in to sharing it.
+#[cfg(target_os = "linux")]
+#[tokio::test]
+async fn test_connected_udp_activates_on_an_adopted_traversal_transport_and_on_a_configured_one() {
+    let mut node_a = make_node();
+    let mut node_b = make_node();
+
+    let transport_id_b = TransportId::new(1);
+    let udp_config = UdpConfig {
+        bind_addr: Some("127.0.0.1:0".to_string()),
+        mtu: Some(1280),
+        ..Default::default()
+    };
+
+    let (packet_tx_a, packet_rx_a) = packet_channel(64);
+    let (packet_tx_b, packet_rx_b) = packet_channel(64);
+
+    node_a.supervisor.packet_tx = Some(packet_tx_a.clone());
+    node_a.packet_rx = Some(packet_rx_a);
+    node_a.supervisor.state = NodeState::Running;
+
+    let mut transport_b = UdpTransport::new(transport_id_b, None, udp_config, packet_tx_b.clone());
+    transport_b.start_async().await.unwrap();
+
+    let addr_b = transport_b.local_addr().unwrap();
+    node_b.supervisor.packet_tx = Some(packet_tx_b.clone());
+    node_b.packet_rx = Some(packet_rx_b);
+    node_b.supervisor.state = NodeState::Running;
+    node_b
+        .transports
+        .insert(transport_id_b, TransportHandle::Udp(transport_b));
+
+    let adopted_socket = std::net::UdpSocket::bind("127.0.0.1:0").unwrap();
+    let handoff =
+        EstablishedTraversal::new("sess-connected", node_b.npub(), addr_b, adopted_socket)
+            .with_transport_name("nostr-punched");
+
+    let result = node_a.adopt_established_traversal(handoff).await.unwrap();
+
+    tokio::select! {
+        result = node_b.run_rx_loop() => {
+            panic!("node_b rx loop exited unexpectedly: {:?}", result);
+        }
+        _ = tokio::time::sleep(Duration::from_millis(500)) => {}
+    }
+
+    tokio::select! {
+        result = node_a.run_rx_loop() => {
+            panic!("node_a rx loop exited unexpectedly: {:?}", result);
+        }
+        _ = tokio::time::sleep(Duration::from_millis(500)) => {}
+    }
+
+    let peer_a_node_addr =
+        *PeerIdentity::from_pubkey_full(node_a.identity().pubkey_full()).node_addr();
+    let peer_b_node_addr =
+        *PeerIdentity::from_pubkey_full(node_b.identity().pubkey_full()).node_addr();
+
+    // Preconditions: both sides are peered, and node_a reaches node_b over
+    // the adopted transport rather than some other one.
+    assert_eq!(node_a.peer_count(), 1, "node_a should promote node_b");
+    assert_eq!(node_b.peer_count(), 1, "node_b should promote node_a");
+    assert!(node_b.get_peer(&peer_a_node_addr).unwrap().has_session());
+    let peer_on_a = node_a.get_peer(&peer_b_node_addr).unwrap();
+    assert!(peer_on_a.has_session());
+    assert_eq!(
+        peer_on_a.transport_id(),
+        Some(result.transport_id),
+        "node_a's peer must be on the adopted transport",
+    );
+    assert!(peer_on_a.current_addr().is_some());
+
+    node_b.activate_connected_udp_sessions().await;
+    node_a.activate_connected_udp_sessions().await;
+
+    assert!(
+        node_b
+            .get_peer(&peer_a_node_addr)
+            .unwrap()
+            .connected_udp()
+            .is_some(),
+        "node_b's peer on its configured listener must get a connected UDP socket; \
+         if it does not, check whether FIPS_CONNECTED_UDP turns the fast path off here",
+    );
+
+    let Some(connected) = node_a.get_peer(&peer_b_node_addr).unwrap().connected_udp() else {
+        let direct =
+            match crate::transport::udp::open_connected_fd(result.local_addr, addr_b, 65536, 65536)
+            {
+                Ok(_) => "succeeds".to_string(),
+                Err(e) => format!("fails with {e}"),
+            };
+        panic!(
+            "node_a's peer on the adopted transport must get a connected UDP socket; \
+             opening one on the adopted socket's address directly {direct}"
+        );
+    };
+    assert_eq!(
+        connected.local_addr(),
+        result.local_addr,
+        "the connected socket must join the adopted socket's port, not another transport's",
+    );
+    drop(connected);
+
+    for (_, transport) in node_a.transports.iter_mut() {
+        transport.stop().await.ok();
+    }
+    for (_, transport) in node_b.transports.iter_mut() {
+        transport.stop().await.ok();
+    }
+}

src/transport/udp/io/mod.rs

@@ -99,6 +99,50 @@ mod tests {
         );
     }
 
+    /// The traversal path binds its socket plainly on port zero, so the
+    /// socket reaches `adopt` carrying neither reuse flag. Adoption has to
+    /// add them after the fact, or the per-peer connected socket's bind to
+    /// the same address is refused with `EADDRINUSE`. Either flag on the
+    /// holder admits that bind on Linux, so the successful open cannot tell
+    /// whether both were set; each flag is also read back.
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn an_adopted_plain_socket_carries_both_reuse_flags_and_admits_a_connected_socket_on_its_port()
+    {
+        use std::os::fd::{AsRawFd, BorrowedFd};
+
+        let peer = std::net::UdpSocket::bind("127.0.0.1:0").expect("failed to bind the peer");
+        let peer_addr = peer.local_addr().expect("peer local address");
+
+        let plain = std::net::UdpSocket::bind(("0.0.0.0", 0)).expect("failed to bind the holder");
+        {
+            let probe = socket2::SockRef::from(&plain);
+            assert!(
+                !probe.reuse_address().expect("read SO_REUSEADDR"),
+                "precondition: a plain bind must arrive without SO_REUSEADDR",
+            );
+            assert!(
+                !probe.reuse_port().expect("read SO_REUSEPORT"),
+                "precondition: a plain bind must arrive without SO_REUSEPORT",
+            );
+        }
+
+        let adopted = UdpRawSocket::adopt(plain, 65536, 65536).expect("failed to adopt the holder");
+
+        let joined = super::open_connected_fd(adopted.local_addr(), peer_addr, 65536, 65536);
+        // SAFETY: `adopted` owns this fd and outlives every use of the borrow.
+        let fd = unsafe { BorrowedFd::borrow_raw(adopted.as_raw_fd()) };
+        let flags = socket2::SockRef::from(&fd);
+        let reuse_address = flags.reuse_address().expect("read SO_REUSEADDR");
+        let reuse_port = flags.reuse_port().expect("read SO_REUSEPORT");
+
+        if let Err(err) = &joined {
+            panic!("a connected socket must be able to bind the adopted socket's port: {err}");
+        }
+        assert!(reuse_address, "the adopted socket must carry SO_REUSEADDR");
+        assert!(reuse_port, "the adopted socket must carry SO_REUSEPORT");
+    }
+
     #[tokio::test]
     async fn test_async_udp_socket_send_recv() {
         let sock1 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536)

src/transport/udp/io/unix.rs

@@ -125,6 +125,8 @@ impl UdpRawSocket {
     /// Adopt an existing bound UDP socket.
     ///
     /// This preserves socket identity/NAT mapping created by bootstrap code.
+    /// The adopted socket is also made joinable by per-peer connected
+    /// sockets, which bind its local address.
     pub fn adopt(
         socket: std::net::UdpSocket,
         recv_buf_size: usize,
@@ -135,6 +137,15 @@ impl UdpRawSocket {
         sock.set_nonblocking(true)
             .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?;
 
+        // A per-peer connected socket later binds this socket's own address,
+        // and the kernel admits that joiner only when the holder carries a
+        // reuse flag too. The socket arrives already bound, so setting the
+        // flags here cannot change which port it was given, unlike flags set
+        // ahead of a port-zero bind; see the comment in `open`. Best-effort,
+        // as there: without them only the connected fast path is refused.
+        let _ = sock.set_reuse_port(true);
+        let _ = sock.set_reuse_address(true);
+
         sock.set_recv_buffer_size(recv_buf_size)
             .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?;
         sock.set_send_buffer_size(send_buf_size)