Merge branch 'master' into next

# Conflicts:
#	docs/reference/configuration.md
#	src/node/handlers/handshake.rs
#	src/node/tests/unit.rs

CHANGELOG.md

@@ -125,6 +125,15 @@ with v0.4.x or earlier peers.
   `LogsDirectory=fips` was added to the packaged systemd units so the capture
   directory is created and cleaned up declaratively.
 
+- `node.rate_limit.established_handshake_burst` and
+  `node.rate_limit.established_handshake_rate`, the parameters of the new
+  established-link msg1 token bucket. Both are optional; omitting them (the
+  normal case) derives the bucket from `node.limits.max_peers`,
+  `node.rekey.after_secs` and `node.rate_limit.handshake_max_resends`, so
+  raising the peer limit sizes the bucket automatically. An explicit zero
+  burst or a non-positive rate is rejected at config validation rather than
+  silently refusing all rekey traffic.
+
 - `packaging/debian/build-deb.sh --features <list>` builds the `.deb` with a
   Cargo feature list, which is how an instrumented package is produced for a
   measurement run. The auto-derived dev Version gains a matching `+<features>`
@@ -195,6 +204,16 @@ with v0.4.x or earlier peers.
   folded into the new tables with a one-time deprecation warning; migrate your
   `fips.yaml` to the new keys.
 
+- Inbound msg1 is classified before it is rate limited, and rekey or restart
+  msg1 arriving on an established link now draws on its own token bucket
+  instead of competing with stranger admission for a single shared one. On a
+  node with many peers the shared bucket refused a large share of ordinary
+  rekey traffic: a field node at roughly 245 peers refused 8753 msg1 in 25
+  minutes, and 159 of the 201 distinct sources were peers it already held
+  sessions with. Nodes upgrade with no config change. The `Msg1 rate limited`
+  log line now reports which limb refused, the pending count or the token
+  bucket, which it previously did not distinguish.
+
 - `SessionDatagram::decrement_ttl` and `SessionDatagram::can_forward` now match
   the forwarder's IP hop-limit semantics: `decrement_ttl` decrements first and
   reports false when the result is zero, and `can_forward` is true only at a

src/node/lifecycle/mod.rs

@@ -1153,8 +1153,20 @@ impl Node {
                         }
                     }
                     match self.adopt_established_traversal(traversal).await {
-                        Ok(_) => {
-                            info!(peer_npub = %peer_npub, "Adopted NAT traversal socket");
+                        Ok(handoff) => {
+                            // `local_addr` lets an operator join a host socket
+                            // table (ss/netstat) against our adoption events.
+                            // `transport_id` separates several peers sharing one
+                            // adopted transport from several adopted transports:
+                            // an adopted transport inherits `accept_connections`,
+                            // so it can admit peers beyond the one it was punched
+                            // for, and the two cases are otherwise identical here.
+                            info!(
+                                peer_npub = %peer_npub,
+                                transport_id = %handoff.transport_id,
+                                local_addr = %handoff.local_addr,
+                                "Adopted NAT traversal socket"
+                            );
                         }
                         Err(err) => {
                             warn!(peer_npub = %peer_npub, error = %err, "Failed to adopt NAT traversal");

src/node/tests/handshake.rs

@@ -1280,11 +1280,22 @@ async fn test_xx_duplicate_msg1_resends_msg2() {
 
 /// `should_admit_msg1` admits when no transport is registered for the id.
 /// (No gate to apply — the caller's other checks decide the outcome.)
+///
+/// This node is also the discriminator for the extraction of
+/// `is_established_link_msg1`: with no transport registered the
+/// `accept_connections` fallback admits, so the two predicates disagree
+/// here and nowhere else. An extraction that dragged the fallback into
+/// `is_established_link_msg1` fails the second assertion.
 #[test]
 fn test_should_admit_msg1_no_transport() {
     let node = make_node();
     let addr = TransportAddr::from_string("10.0.0.2:2121");
     assert!(node.should_admit_msg1(TransportId::new(1), &addr));
+    assert!(
+        !node.is_established_link_msg1(TransportId::new(1), &addr),
+        "the accept_connections fallback must not be part of the \
+         established-link predicate"
+    );
 }
 
 /// `should_admit_msg1` rejects a fresh msg1 (no addr_to_link entry) when
@@ -1459,6 +1470,12 @@ async fn test_should_admit_msg1_admits_rekey_when_addr_form_differs() {
         !node.should_admit_msg1(transport_id, &stranger_addr),
         "fresh msg1 from unknown source must still be rejected"
     );
+
+    // The same two predicates read directly: both addr-forms of the
+    // established peer are established links, the stranger is not.
+    assert!(node.is_established_link_msg1(transport_id, &hostname_addr));
+    assert!(node.is_established_link_msg1(transport_id, &numeric_addr));
+    assert!(!node.is_established_link_msg1(transport_id, &stranger_addr));
 }
 
 /// `is_established_link_msg1` and `should_admit_msg1` are deliberately

src/peer/connected_udp/drain.rs

@@ -1,8 +1,3 @@
-// Paired with the connected-socket opener (`io::open_connected_fd`):
-// dormant in this PR until the activation handler is wired into the
-// node tick (follow-up).
-#![allow(dead_code)]
-
 //! Recv-side drain thread for a per-peer connected UDP socket.
 //!
 //! Once a UDP socket is `connect()`-ed to a peer, Linux and Darwin

src/peer/connected_udp/socket.rs

@@ -3,7 +3,6 @@
 //! Adopts an fd produced by `crate::transport::udp::open_connected_fd`
 //! and closes it on drop. See that function's docs for why established
 //! peers get their own connected socket.
-#![allow(dead_code)]
 
 use std::net::SocketAddr;
 use std::os::unix::io::{AsRawFd, OwnedFd, RawFd};

src/transport/udp/io.rs

@@ -803,11 +803,6 @@ pub use platform::{AsyncUdpSocket, UdpRawSocket};
 /// is libc-syscall + `sockopts_macos` specific.
 #[cfg(any(target_os = "linux", target_os = "macos"))]
 mod connected {
-    // The connected-UDP fast path is infra-ready but not yet wired into
-    // the encrypt-worker dispatch site (a follow-up PR will refcount-clone
-    // the socket into each FmpSendJob). Keep the API surface in tree.
-    #![allow(dead_code)]
-
     use std::io;
     use std::net::SocketAddr;
     use std::os::unix::io::{AsRawFd, FromRawFd, OwnedFd, RawFd};
@@ -900,7 +895,7 @@ mod connected {
             )
         };
         if bind_r < 0 {
-            return Err(io::Error::last_os_error());
+            return Err(syscall_err("bind", local_addr));
         }
 
         // Connect to the peer — locks in the per-packet kernel route.
@@ -913,12 +908,17 @@ mod connected {
             )
         };
         if conn_r < 0 {
-            return Err(io::Error::last_os_error());
+            return Err(syscall_err("connect", peer_addr));
         }
 
         Ok(owned)
     }
 
+    fn syscall_err(syscall: &str, addr: SocketAddr) -> io::Error {
+        let err = io::Error::last_os_error();
+        io::Error::new(err.kind(), format!("{syscall} {addr}: {err}"))
+    }
+
     #[cfg(not(target_os = "linux"))]
     fn set_nonblocking_cloexec(fd: RawFd) -> io::Result<()> {
         let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
@@ -1005,6 +1005,50 @@ mod connected {
             )
         };
     }
+
+    #[cfg(all(test, target_os = "linux"))]
+    mod tests {
+        use super::*;
+        use std::net::UdpSocket;
+
+        const BUF: usize = 1 << 20;
+
+        #[test]
+        fn bind_failure_names_bind_and_local_addr() {
+            // A plain socket without SO_REUSEPORT holds the address, so the
+            // SO_REUSEPORT bind below is refused with EADDRINUSE.
+            let holder = UdpSocket::bind("127.0.0.1:0").expect("holder bind");
+            let holder_addr = holder.local_addr().expect("holder addr");
+
+            let err = open_connected_fd(holder_addr, "127.0.0.1:9".parse().unwrap(), BUF, BUF)
+                .expect_err("bind must fail against a non-reuseport holder");
+
+            assert_eq!(err.kind(), io::ErrorKind::AddrInUse, "{err}");
+            let msg = err.to_string();
+            assert!(msg.starts_with("bind "), "{msg}");
+            assert!(msg.contains(&holder_addr.to_string()), "{msg}");
+            assert!(!msg.contains("connect"), "{msg}");
+        }
+
+        #[test]
+        fn connect_failure_names_connect_and_peer_addr() {
+            // connect(2) to the broadcast address without SO_BROADCAST fails
+            // synchronously with EACCES; the bind before it succeeds.
+            let err = open_connected_fd(
+                "127.0.0.1:0".parse().unwrap(),
+                "255.255.255.255:9999".parse().unwrap(),
+                BUF,
+                BUF,
+            )
+            .expect_err("connect to broadcast without SO_BROADCAST must fail");
+
+            assert_eq!(err.kind(), io::ErrorKind::PermissionDenied, "{err}");
+            let msg = err.to_string();
+            assert!(msg.starts_with("connect "), "{msg}");
+            assert!(msg.contains("255.255.255.255:9999"), "{msg}");
+            assert!(!msg.contains("bind"), "{msg}");
+        }
+    }
 }
 
 #[cfg(any(target_os = "linux", target_os = "macos"))]

src/transport/udp/sockopts_macos.rs

@@ -1,5 +1,6 @@
-// Applied inside `io::open_connected_fd` (see `io.rs`), dormant until
-// the connected-UDP fast path is wired into dispatch — hence the allow.
+// Applied inside `io::open_connected_fd` (see `io.rs`). The full
+// `NET_SERVICE_TYPE_*` table is kept for reference although only `OAM` is
+// selected, so most of these constants are deliberately unreferenced.
 #![allow(dead_code)]
 
 //! Darwin UDP socket tuning.