fix(udp): size and align control message buffers from their payloads

.github/workflows/ci.yml

@@ -64,6 +64,48 @@ jobs:
       - name: Build and test noq-udp (posix_minimal)
         run: cargo test --locked -p noq-udp
 
+  test_musl:
+    name: Test ${{ matrix.target }}
+    if: "github.event_name != 'pull_request' || ! contains(github.event.pull_request.labels.*.name, 'flaky-test')"
+    timeout-minutes: 30
+    strategy:
+      fail-fast: false
+      matrix:
+        include:
+          # musl binaries are statically linked, and each target runs on a runner
+          # of its own architecture, so the tests run natively.
+          - target: aarch64-unknown-linux-musl
+            runner: ubuntu-24.04-arm
+          - target: x86_64-unknown-linux-musl
+            runner: ubuntu-latest
+    runs-on: ${{ matrix.runner }}
+    env:
+      RUSTC_WRAPPER: "sccache"
+      SCCACHE_GHA_ENABLED: "on"
+      # When cross-compiling to musl, cc-rs looks for `<arch>-linux-musl-gcc`,
+      # but `musl-tools` only ships the host-native `musl-gcc`.
+      CC_aarch64_unknown_linux_musl: musl-gcc
+      CC_x86_64_unknown_linux_musl: musl-gcc
+    steps:
+      - uses: actions/checkout@v6
+      - uses: dtolnay/rust-toolchain@stable
+        with:
+          target: ${{ matrix.target }}
+      - uses: mozilla-actions/sccache-action@v0.0.9
+      - name: Install musl toolchain
+        run: |
+          sudo apt-get update
+          sudo apt-get install -y musl-tools
+      - name: Install cargo-nextest
+        uses: taiki-e/install-action@v2
+        with:
+          tool: nextest@0.9.80
+      - name: Run tests
+        run: |
+          cargo nextest run --locked --workspace --exclude fuzz --lib --bins --tests --target ${{ matrix.target }} --profile ci
+        env:
+          RUST_LOG: ${{ runner.debug && 'TRACE' || 'DEBUG'}}
+
   esp32_check:
     name: ESP32-C3 build check (noq-udp, noq-proto, noq)
     if: "github.event_name != 'pull_request' || ! contains(github.event.pull_request.labels.*.name, 'flaky-test')"

noq-udp/src/apple_fast.rs

@@ -33,7 +33,7 @@ fn send_via_sendmsg_x(
 ) -> io::Result<()> {
     let mut hdrs = unsafe { mem::zeroed::<[msghdr_x; BATCH_SIZE]>() };
     let mut iovs = unsafe { mem::zeroed::<[libc::iovec; BATCH_SIZE]>() };
-    let mut ctrls = [cmsg::Aligned([0u8; cmsg::LEN]); BATCH_SIZE];
+    let mut ctrls = [cmsg::SendBuf::zeroed(); BATCH_SIZE];
     let addr = socket2::SockAddr::from(transmit.destination);
     let segment_size = transmit.segment_size.unwrap_or(transmit.contents.len());
     let mut cnt = 0;
@@ -75,7 +75,7 @@ fn prepare_msg_x(
     dst_addr: &socket2::SockAddr,
     hdr: &mut msghdr_x,
     iov: &mut libc::iovec,
-    ctrl: &mut cmsg::Aligned<[u8; cmsg::LEN]>,
+    ctrl: &mut cmsg::SendBuf,
     #[allow(unused_variables)] encode_src_ip: bool,
     sendmsg_einval: bool,
 ) {
@@ -89,18 +89,18 @@ fn prepare_msg_x(
     hdr.msg_iov = iov;
     hdr.msg_iovlen = 1;
 
-    hdr.msg_control = ctrl.0.as_mut_ptr() as _;
-    hdr.msg_controllen = cmsg::LEN as _;
+    hdr.msg_control = ctrl.as_mut_ptr() as _;
+    hdr.msg_controllen = ctrl.len() as _;
     let mut encoder = unsafe { cmsg::Encoder::new(hdr) };
     let ecn = transmit.ecn.map_or(0, |x| x as libc::c_int);
     let is_ipv4 = transmit.destination.is_ipv4()
         || matches!(transmit.destination.ip(), IpAddr::V6(addr) if addr.to_ipv4_mapped().is_some());
     if is_ipv4 {
         if !sendmsg_einval {
-            encoder.push(libc::IPPROTO_IP, libc::IP_TOS, ecn as IpTosTy);
+            encoder.push_ecn_v4(ecn as IpTosTy);
         }
     } else {
-        encoder.push(libc::IPPROTO_IPV6, libc::IPV6_TCLASS, ecn);
+        encoder.push_ecn_v6(ecn);
     }
 
     if let Some(ip) = &transmit.src_ip {
@@ -110,7 +110,7 @@ fn prepare_msg_x(
                     let addr = libc::in_addr {
                         s_addr: u32::from_ne_bytes(v4.octets()),
                     };
-                    encoder.push(libc::IPPROTO_IP, libc::IP_RECVDSTADDR, addr);
+                    encoder.push_src_addr_v4(addr);
                 }
             }
             IpAddr::V6(v6) => {
@@ -120,7 +120,7 @@ fn prepare_msg_x(
                         s6_addr: v6.octets(),
                     },
                 };
-                encoder.push(libc::IPPROTO_IPV6, libc::IPV6_PKTINFO, pktinfo);
+                encoder.push_pktinfo_v6(pktinfo);
             }
         }
     }
@@ -156,7 +156,7 @@ pub(crate) fn recv_via_recvmsg_x(
     // uninitialized memory, do not use `MaybeUninit` for `ctrls`, instead
     // initialize `ctrls` with `0`s. A control message of all `0`s is
     // automatically skipped by `libc::CMSG_NXTHDR`.
-    let mut ctrls = [cmsg::Aligned([0u8; cmsg::LEN]); BATCH_SIZE];
+    let mut ctrls = [cmsg::RecvBuf::zeroed(); BATCH_SIZE];
     let mut hdrs = unsafe { mem::zeroed::<[msghdr_x; BATCH_SIZE]>() };
     let max_msg_count = bufs.len().min(BATCH_SIZE);
     for i in 0..max_msg_count {
@@ -178,15 +178,15 @@ pub(crate) fn recv_via_recvmsg_x(
 fn prepare_recv_x(
     buf: &mut IoSliceMut<'_>,
     name: &mut MaybeUninit<libc::sockaddr_storage>,
-    ctrl: &mut cmsg::Aligned<[u8; cmsg::LEN]>,
+    ctrl: &mut cmsg::RecvBuf,
     hdr: &mut msghdr_x,
 ) {
     hdr.msg_name = name.as_mut_ptr() as _;
     hdr.msg_namelen = size_of::<libc::sockaddr_storage>() as _;
     hdr.msg_iov = buf as *mut IoSliceMut<'_> as *mut libc::iovec;
     hdr.msg_iovlen = 1;
-    hdr.msg_control = ctrl.0.as_mut_ptr() as _;
-    hdr.msg_controllen = cmsg::LEN as _;
+    hdr.msg_control = ctrl.as_mut_ptr() as _;
+    hdr.msg_controllen = ctrl.len() as _;
     hdr.msg_flags = 0;
     hdr.msg_datalen = buf.len();
 }
@@ -248,4 +248,8 @@ impl MsgHdr for msghdr_x {
     fn control_len(&self) -> usize {
         self.msg_controllen as _
     }
+
+    fn recv_flags(&self) -> libc::c_int {
+        self.msg_flags
+    }
 }

noq-udp/src/cmsg/mod.rs

@@ -1,6 +1,7 @@
 use std::{
     ffi::{c_int, c_uchar},
     ptr,
+    sync::atomic::{AtomicBool, Ordering},
 };
 
 #[cfg(unix)]
@@ -11,7 +12,7 @@ mod imp;
 #[path = "windows.rs"]
 mod imp;
 
-pub(crate) use imp::Aligned;
+pub(crate) use imp::{PAYLOAD_ALIGN, RecvBuf, SendBuf};
 
 /// Helper to encode a series of control messages (native "cmsgs") to a buffer for use in `sendmsg`
 //  like API.
@@ -39,11 +40,17 @@ impl<'a, M: MsgHdr> Encoder<'a, M> {
 
     /// Append a control message to the buffer.
     ///
+    /// Private: each message we send has its own method, next to the size covering it.
+    ///
     /// # Panics
     /// - If insufficient buffer space remains.
-    /// - If `T` has stricter alignment requirements than `M::ControlMessage`
-    pub(crate) fn push<T: Copy>(&mut self, level: c_int, ty: c_int, value: T) {
-        assert!(align_of::<T>() <= align_of::<M::ControlMessage>());
+    fn push<T: Copy>(&mut self, level: c_int, ty: c_int, value: T) {
+        const {
+            assert!(
+                align_of::<T>() <= PAYLOAD_ALIGN,
+                "control message payload is more aligned than a control message buffer can be",
+            );
+        }
         let space = M::ControlMessage::cmsg_space(size_of_val(&value));
         assert!(
             self.hdr.control_len() >= self.len + space,
@@ -74,11 +81,31 @@ impl<M: MsgHdr> Drop for Encoder<'_, M> {
     }
 }
 
+/// Warns once if the kernel had more to say about a datagram than the buffer could hold.
+///
+/// The dropped messages cost us metadata, at worst the GRO segment size. `RECV_LEN` covers
+/// every option we enable, so one is unaccounted for, or the caller enabled their own on
+/// the socket they gave us.
+pub(crate) fn warn_if_control_truncated(hdr: &impl MsgHdr) {
+    static WARNED: AtomicBool = AtomicBool::new(false);
+
+    if hdr.recv_flags() & imp::MSG_CTRUNC != 0 && !WARNED.swap(true, Ordering::Relaxed) {
+        crate::log::warn!(
+            "control messages truncated on receive, some datagram metadata was dropped"
+        );
+    }
+}
+
 /// # Safety
 ///
 /// `cmsg` must refer to a native cmsg containing a payload of type `T`
 pub(crate) unsafe fn decode<T: Copy, C: CMsgHdr>(cmsg: &impl CMsgHdr) -> T {
-    assert!(align_of::<T>() <= align_of::<C>());
+    const {
+        assert!(
+            align_of::<T>() <= PAYLOAD_ALIGN,
+            "control message payload is more aligned than a control message buffer can be",
+        );
+    }
     debug_assert_eq!(cmsg.len(), C::cmsg_len(size_of::<T>()));
     unsafe { ptr::read(cmsg.cmsg_data() as *const T) }
 }
@@ -138,6 +165,9 @@ pub(crate) trait MsgHdr {
     fn set_control_len(&mut self, len: usize);
 
     fn control_len(&self) -> usize;
+
+    /// The flags the kernel set on a received message, i.e. `msg_flags`.
+    fn recv_flags(&self) -> c_int;
 }
 
 pub(crate) trait CMsgHdr {
@@ -151,6 +181,3 @@ pub(crate) trait CMsgHdr {
 
     fn len(&self) -> usize;
 }
-
-#[cfg(unix)]
-pub(crate) const LEN: usize = 96;