diff --git a/src/node/tests/unit.rs b/src/node/tests/unit.rs index 044d014a..f26cdbd0 100644 --- a/src/node/tests/unit.rs +++ b/src/node/tests/unit.rs @@ -3386,8 +3386,8 @@ async fn app_owned_udp_fd_seam_stays_silent_without_a_udp_transport() { } /// A UDP transport that never bound has no fd to hand out. The bind address is -/// deliberately unparseable — a busy port would not do it, since -/// `UdpRawSocket::open` sets `SO_REUSEADDR`/`SO_REUSEPORT` before binding. +/// deliberately unparseable, so the failure is in parsing and cannot depend on +/// what else happens to hold a port while the suite runs. #[cfg(unix)] #[tokio::test] async fn app_owned_udp_fd_seam_stays_silent_when_the_udp_transport_fails_to_start() { diff --git a/src/transport/udp/io/mod.rs b/src/transport/udp/io/mod.rs index e372afe7..8acbb630 100644 --- a/src/transport/udp/io/mod.rs +++ b/src/transport/udp/io/mod.rs @@ -76,6 +76,29 @@ mod tests { assert!(send_buf > 0, "send buffer should be non-zero"); } + /// The listen socket's reuse flags go on after its own bind, so they + /// never license the kernel to hand this socket a port someone else + /// holds. The visible consequence is that a second bind of an occupied + /// port fails loudly instead of silently sharing it and splitting the + /// inbound datagrams between the two recv loops. + #[cfg(unix)] + #[test] + fn a_second_open_of_an_occupied_port_fails_instead_of_sharing_it() { + let holder = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) + .expect("failed to bind the holding socket"); + let addr = holder.local_addr(); + + // `UdpRawSocket` is not `Debug`, so this cannot be `expect_err`. + let Err(err) = UdpRawSocket::open(addr, 65536, 65536) else { + panic!("the port is already held, so the second bind must fail"); + }; + + assert!( + err.to_string().contains("bind failed"), + "the failure must come from bind, not from a later step: {err}", + ); + } + #[tokio::test] async fn test_async_udp_socket_send_recv() { let sock1 = UdpRawSocket::open("127.0.0.1:0".parse().unwrap(), 65536, 65536) diff --git a/src/transport/udp/io/unix.rs b/src/transport/udp/io/unix.rs index 66e8e5e3..60a4cb43 100644 --- a/src/transport/udp/io/unix.rs +++ b/src/transport/udp/io/unix.rs @@ -57,19 +57,27 @@ impl UdpRawSocket { sock.set_nonblocking(true) .map_err(|e| TransportError::StartFailed(format!("set nonblocking failed: {}", e)))?; - // SO_REUSEPORT lets per-peer `ConnectedPeerSocket`s bind - // to the same wildcard port the listen socket holds. Must - // be set BEFORE bind. Without this, the connected-UDP - // activation handler fails with EADDRINUSE on Linux and - // every outbound packet falls back to the wildcard listen - // socket — losing the kernel 5-tuple cache benefit and - // most of the multihop forwarding throughput gain. - let _ = sock.set_reuse_port(true); - let _ = sock.set_reuse_address(true); - sock.bind(&bind_addr.into()) .map_err(|e| TransportError::StartFailed(format!("bind failed: {}", e)))?; + // SO_REUSEPORT lets per-peer `ConnectedPeerSocket`s bind to the same + // port this listen socket holds; without it the connected-UDP + // activation handler fails with EADDRINUSE on Linux and every + // outbound packet falls back to the wildcard listen socket, losing + // the kernel 5-tuple cache benefit and most of the multihop + // forwarding throughput gain. + // + // Set AFTER bind, deliberately. The flags mean different things + // either side of it: before bind, "the kernel may give me a port + // another socket already holds", which for the `outbound_only` + // `0.0.0.0:0` bind means duplicate ephemeral ports, and for a + // configured port means a second daemon silently shares it instead + // of failing to start; after bind, "another socket may later join my + // port", which is the only half the fast path needs. A joiner still + // binds successfully against a holder flagged after its own bind. + let _ = sock.set_reuse_port(true); + let _ = sock.set_reuse_address(true); + // Set socket buffer sizes sock.set_recv_buffer_size(recv_buf_size) .map_err(|e| TransportError::StartFailed(format!("set recv buffer: {}", e)))?;