syd 3.54.1

// Syd: rock-solid application kernel
// src/kernel/net/recvmsg.rs: recvmsg(2), recvmmsg(2), and recvmmsg_time64 handlers
//
// Copyright (c) 2025, 2026 Ali Polatel <alip@chesswob.org>
//
// SPDX-License-Identifier: GPL-3.0

use std::{io::IoSlice, os::fd::AsFd};

use libc::{c_int, c_uint, c_void, iovec, sockaddr, socklen_t, MSG_CTRUNC};
use libseccomp::ScmpNotifResp;
use nix::{
    errno::Errno,
    sys::{
        socket::{SockaddrLike, SockaddrStorage},
        uio::RemoteIoVec,
    },
};
use zeroize::Zeroizing;

use crate::{
    compat::{
        mmsghdr, mmsghdr32, msghdr, msghdr32, recvmmsg, recvmsg, try_from_bytes, AddressFamily,
        MmsgHdr, MsgFlags, TimeSpec32, TimeSpec64, ToByteArray, UIO_MAXIOV,
    },
    confine::scmp_arch_is_compat32,
    fd::{fd_inode, has_recv_timeout, SafeOwnedFd},
    kernel::net::{to_msgflags, SockOpts},
    req::UNotifyEventRequest,
    unix::unix_addr_len,
};

const SOCKADDR_SIZE: usize = size_of::<libc::sockaddr_storage>();

pub(crate) fn handle_recvmsg(
    fd: SafeOwnedFd,
    request: &UNotifyEventRequest,
    args: &[u64; 6],
    opts: SockOpts,
) -> Result<ScmpNotifResp, Errno> {
    let SockOpts {
        sock_dom,
        flags,
        options,
        is_nonblock,
    } = opts;

    // Truncate flags to 32-bit keeping unknown flags.
    let call_flags = to_msgflags(args[2]);

    // Reject MSG_OOB as necessary.
    if !options.allow_unsafe_oob() && call_flags.contains(MsgFlags::MSG_OOB) {
        // Signal no support to let the sandbox process handle the error
        // gracefully. This is consistent with the Linux kernel.
        return Err(Errno::EOPNOTSUPP);
    }

    // Determine bitness of sandbox process.
    let req = request.scmpreq;
    let is32 = scmp_arch_is_compat32(req.data.arch);

    // Linux rejects MSG_CMSG_COMPAT on 64-bit.
    if !is32 && call_flags.contains(MsgFlags::MSG_CMSG_COMPAT) {
        return Err(Errno::EINVAL);
    }

    // Read user msghdr.
    let hdr_sz = if is32 {
        size_of::<msghdr32>()
    } else {
        size_of::<msghdr>()
    };

    let hdr = request.read_vec_all_zeroed(args[1], hdr_sz)?;
    let mut hdr: msghdr = if is32 {
        let m32: msghdr32 = try_from_bytes(&hdr)?;
        msghdr::from(m32)
    } else {
        try_from_bytes(&hdr)?
    };

    // Mirror sandbox process iovecs to local, bounded buffers.
    #[expect(clippy::type_complexity)]
    let mut msg_bufs: Vec<(Zeroizing<Vec<u8>>, u64)> = Vec::new();
    let mut msg_iovs: Vec<iovec> = Vec::new();
    let mut nam_buf: Vec<u8> = Vec::new();
    let mut ctl_buf: Vec<u8> = Vec::new();

    // Handle msg_name.
    let (user_nam_base, user_nam_size) = request.setup_msghdr_name(&mut hdr, &mut nam_buf)?;

    // Handle msg_iov.
    let user_iov_base = request.read_msghdr_iov(&mut hdr, &mut msg_bufs, &mut msg_iovs)?;

    // Handle msg_control.
    let (user_ctl_base, user_ctl_size) = request.setup_msghdr_ctl(&mut hdr, &mut ctl_buf)?;

    // Handle scatter buffers to be used post-syscall.
    //
    // Buffer consists of message payload, control message, header, and address.
    let buf_len = msg_bufs.len().checked_add(3).ok_or(Errno::EOVERFLOW)?;
    let mut iovs_l: Vec<IoSlice<'_>> = Vec::new();
    let mut iovs_r: Vec<RemoteIoVec> = Vec::new();
    iovs_l.try_reserve(buf_len).or(Err(Errno::ENOMEM))?;
    iovs_r.try_reserve(buf_len).or(Err(Errno::ENOMEM))?;

    let mmsghdr_size = if is32 {
        size_of::<msghdr32>()
    } else {
        size_of::<msghdr>()
    };
    let mut hdr_buf: Zeroizing<Vec<u8>> = Zeroizing::new(Vec::new());
    hdr_buf.try_reserve(mmsghdr_size).or(Err(Errno::ENOMEM))?;
    hdr_buf.resize(mmsghdr_size, 0);

    // Handle address buffer.
    let mut addr_buf: Zeroizing<[u8; SOCKADDR_SIZE]> = Zeroizing::new([0u8; SOCKADDR_SIZE]);

    // Track blocking call for invalidation semantics.
    let is_blocking = !is_nonblock && !call_flags.contains(MsgFlags::MSG_DONTWAIT);
    let ignore_restart = if is_blocking {
        has_recv_timeout(&fd)?
    } else {
        false
    };
    if is_blocking {
        request.cache.add_sys_block(req, ignore_restart)?;
    }

    // Perform recvmsg(2).
    let result = recvmsg(&fd, hdr.as_mut(), call_flags);

    // Remove invalidation record.
    if is_blocking {
        request.cache.del_sys_block(req.id)?;
    }

    // Check result after critical block.
    let r_bytes = result?.bytes;

    // Scatter payload into sandbox process iov buffers.
    scatter_iov(r_bytes, &msg_bufs, &mut iovs_l, &mut iovs_r)?;

    // Handle peer address logic.
    //
    // Linux rejects negative values for msg_namelen.
    let namelen: socklen_t = socklen_t::try_from(hdr.msg_namelen).or(Err(Errno::EINVAL))?;
    let (namelen_out, addr_len) = if sock_dom == AddressFamily::Unix {
        fixup_unix_addr(&fd, request, hdr.msg_name, namelen, &mut *addr_buf)?
    } else {
        copy_addr(hdr.msg_name, namelen, &mut *addr_buf)?
    };
    hdr.msg_namelen = c_int::try_from(namelen_out).or(Err(Errno::EINVAL))?;

    // Handle control messages.
    //
    // Pass unsupported control messages unchanged.
    let cmsg_out = if !hdr.msg_control.is_null() && hdr.msg_controllen > 0 {
        // SAFETY: msg_control points to a locally allocated buffer checked above.
        let cmsg_buf =
            unsafe { std::slice::from_raw_parts(hdr.msg_control as *const u8, hdr.msg_controllen) };
        let close_on_exec =
            flags.force_cloexec() || call_flags.contains(MsgFlags::MSG_CMSG_CLOEXEC);
        let rand_fd = flags.force_rand_fd();

        let (cmsgs, cmsgs_truncated) =
            request.fixup_cmsgs(&fd, cmsg_buf, user_ctl_size, close_on_exec, rand_fd)?;
        let (out_buf, cmsg_len, truncated) = request.setup_cmsgs(&cmsgs, user_ctl_size)?;

        if truncated || cmsgs_truncated {
            hdr.msg_flags |= MSG_CTRUNC as c_uint;
        }
        hdr.msg_controllen = cmsg_len;

        Some(out_buf)
    } else {
        hdr.msg_controllen = 0;

        None
    };

    // Copy message header into pre-allocated buffer.
    //
    // Replace local pointers with sandbox process pointers.
    hdr.msg_iov = user_iov_base as *mut iovec;
    hdr.msg_name = user_nam_base as *mut c_void;
    hdr.msg_control = user_ctl_base as *mut c_void;

    // Copy message header handling 32-bit as necessary.
    if is32 {
        let m32: msghdr32 = hdr.try_into()?;
        let buf: [u8; size_of::<msghdr32>()] = m32.to_byte_array();
        hdr_buf.copy_from_slice(&buf);
    } else {
        let buf: [u8; size_of::<msghdr>()] = hdr.to_byte_array();
        hdr_buf.copy_from_slice(&buf);
    }

    // Gather control message, header, and address into the batch.
    //
    // Gather control message.
    if let Some(ref out_buf) = cmsg_out {
        let cmsg_len = hdr.msg_controllen;
        if cmsg_len > 0 {
            iovs_l.push(IoSlice::new(&out_buf[..cmsg_len]));
            iovs_r.push(RemoteIoVec {
                base: usize::try_from(user_ctl_base).or(Err(Errno::EOVERFLOW))?,
                len: cmsg_len,
            });
        }
    }

    // Gather message header.
    iovs_l.push(IoSlice::new(&hdr_buf));
    iovs_r.push(RemoteIoVec {
        base: usize::try_from(args[1]).or(Err(Errno::EOVERFLOW))?,
        len: hdr_buf.len(),
    });

    // Gather peer address.
    #[expect(clippy::cast_possible_truncation)]
    let out_len = (namelen_out.min(user_nam_size as socklen_t)) as usize;
    let out_len = out_len.min(addr_len);
    if out_len > 0 {
        iovs_l.push(IoSlice::new(&addr_buf[..out_len]));
        iovs_r.push(RemoteIoVec {
            base: usize::try_from(user_nam_base).or(Err(Errno::EOVERFLOW))?,
            len: out_len,
        });
    }

    // Write in single batch.
    //
    // Linux rejects copy failures with EFAULT.
    if !iovs_l.is_empty() {
        let siz: usize = iovs_r.iter().map(|v| v.len).sum();
        let len = request.write_mem_many_all(&iovs_l, &iovs_r)?;
        if len != siz {
            return Err(Errno::EFAULT);
        }
    }

    // Return number of payload bytes received.
    #[expect(clippy::cast_possible_wrap)]
    Ok(request.return_syscall(r_bytes as i64))
}

pub(crate) fn handle_recvmmsg(
    fd: SafeOwnedFd,
    request: &UNotifyEventRequest,
    args: &[u64; 6],
    opts: SockOpts,
) -> Result<ScmpNotifResp, Errno> {
    // Determine if the process is 32-bit or 64-bit.
    let is32 = scmp_arch_is_compat32(request.scmpreq.data.arch);

    // Read the timespec structure for timeout (32-bit or 64-bit).
    let timeout = if args[4] != 0 {
        if is32 {
            // Read TimeSpec32 if the process is 32-bit.
            Some(request.remote_timespec32(args[4])?)
        } else {
            // Read TimeSpec64 if the process is 64-bit.
            Some(request.remote_timespec64(args[4])?)
        }
    } else {
        None
    };

    // Pass the timeout to the internal function.
    do_recvmmsg(fd, args, request, opts, timeout, is32)
}

pub(crate) fn handle_recvmmsg64(
    fd: SafeOwnedFd,
    request: &UNotifyEventRequest,
    args: &[u64; 6],
    opts: SockOpts,
) -> Result<ScmpNotifResp, Errno> {
    // Read the timespec structure for timeout (explicit 64-bit).
    let timeout = if args[4] != 0 {
        Some(request.remote_timespec64(args[4])?)
    } else {
        None
    };

    // Pass the timeout to the internal function.
    do_recvmmsg(fd, args, request, opts, timeout, false /*is32*/)
}

// Helper to handle both recvmmsg(2) and recvmmsg_time64(2) syscalls.
#[expect(clippy::cognitive_complexity)]
fn do_recvmmsg<Fd: AsFd>(
    fd: Fd,
    args: &[u64; 6],
    request: &UNotifyEventRequest,
    opts: SockOpts,
    mut timeout: Option<TimeSpec64>,
    timeout_is32: bool,
) -> Result<ScmpNotifResp, Errno> {
    let SockOpts {
        sock_dom,
        flags,
        options,
        is_nonblock,
    } = opts;

    // Truncate flags to 32-bit keeping unknown flags.
    let call_flags = to_msgflags(args[3]);

    // Reject MSG_OOB as necessary.
    if !options.allow_unsafe_oob() && call_flags.contains(MsgFlags::MSG_OOB) {
        // Signal no support to let the sandbox process handle the error
        // gracefully. This is consistent with the Linux kernel.
        return Err(Errno::EOPNOTSUPP);
    }

    // Determine bitness of sandbox process.
    let req = request.scmpreq;
    let is32 = scmp_arch_is_compat32(req.data.arch);

    // Linux rejects MSG_CMSG_COMPAT on 64-bit.
    if !is32 && call_flags.contains(MsgFlags::MSG_CMSG_COMPAT) {
        return Err(Errno::EINVAL);
    }

    // Read the user mmsghdr array.
    //
    // Linux truncates message count to unsigned int. Cap at IOV_MAX.
    #[expect(clippy::cast_possible_truncation)]
    let msg_count = (args[2] as c_uint as usize).min(UIO_MAXIOV);
    let msgs_offset = args[1];

    // Preallocate memory for mmsghdr array.
    let hdr_sz = if is32 {
        size_of::<mmsghdr32>()
    } else {
        size_of::<mmsghdr>()
    };
    let total_sz = hdr_sz.checked_mul(msg_count).ok_or(Errno::EOVERFLOW)?;
    let hdr = request.read_vec_all_zeroed(msgs_offset, total_sz)?;

    // Convert to native format to pass to recvmmsg(2).
    let mut msgs = Vec::new();
    #[expect(clippy::type_complexity)]
    let mut msg_bufs: Vec<Option<Vec<(Zeroizing<Vec<u8>>, u64)>>> = Vec::new();
    let mut nam_bufs: Vec<Option<Vec<u8>>> = Vec::new();
    let mut ctl_bufs: Vec<Option<Vec<u8>>> = Vec::new();
    let mut msg_iovs: Vec<Vec<iovec>> = Vec::new();
    let mut user_iov_bases: Vec<Option<u64>> = Vec::new();
    let mut user_nam_bases: Vec<Option<(u64, usize)>> = Vec::new();
    let mut user_ctl_bases: Vec<Option<(u64, usize)>> = Vec::new();
    msgs.try_reserve(msg_count).or(Err(Errno::ENOMEM))?;
    msg_bufs.try_reserve(msg_count).or(Err(Errno::ENOMEM))?;
    nam_bufs.try_reserve(msg_count).or(Err(Errno::ENOMEM))?;
    ctl_bufs.try_reserve(msg_count).or(Err(Errno::ENOMEM))?;
    msg_iovs.try_reserve(msg_count).or(Err(Errno::ENOMEM))?;
    user_iov_bases
        .try_reserve(msg_count)
        .or(Err(Errno::ENOMEM))?;
    user_nam_bases
        .try_reserve(msg_count)
        .or(Err(Errno::ENOMEM))?;
    user_ctl_bases
        .try_reserve(msg_count)
        .or(Err(Errno::ENOMEM))?;

    for chunk in hdr.chunks(hdr_sz) {
        let inner: libc::mmsghdr = if is32 {
            let m32: mmsghdr32 = try_from_bytes(chunk)?;
            mmsghdr::from(m32).into()
        } else {
            let m64: mmsghdr = try_from_bytes(chunk)?;
            m64.into()
        };
        msgs.push(MmsgHdr::from_raw(inner));
    }

    // Read all iov arrays in one batch.
    request.read_mmsghdr_iovs(&mut msgs, &mut msg_bufs, &mut msg_iovs, &mut user_iov_bases)?;

    // Set up name and control buffers.
    for mmhdr in &mut msgs {
        request.setup_mmsghdr_name(mmhdr.as_inner_mut(), &mut nam_bufs, &mut user_nam_bases)?;
        request.setup_mmsghdr_ctl(mmhdr.as_inner_mut(), &mut ctl_bufs, &mut user_ctl_bases)?;
    }

    // Handle scatter buffers to be used post-syscall.
    //
    // Buffer consists of the following items:
    // 1. Payload iov total
    // 2. Control message + header + address: +3 per message
    // 3. Timeout.
    let buf_len: usize = msg_bufs
        .iter()
        .filter_map(Option::as_ref)
        .map(Vec::len)
        .try_fold(0usize, |acc, n| acc.checked_add(n))
        .ok_or(Errno::EOVERFLOW)?;
    let buf_len = msg_count
        .checked_mul(3)
        .and_then(|n| n.checked_add(buf_len))
        .and_then(|n| n.checked_add(1)) // timeout
        .ok_or(Errno::EOVERFLOW)?;

    let mut iovs_l: Vec<IoSlice<'_>> = Vec::new();
    let mut iovs_r: Vec<RemoteIoVec> = Vec::new();
    iovs_l.try_reserve(buf_len).or(Err(Errno::ENOMEM))?;
    iovs_r.try_reserve(buf_len).or(Err(Errno::ENOMEM))?;

    // Allocate per-message data.
    let mut cmsg_outs: Vec<Option<Zeroizing<Vec<u8>>>> = Vec::new();
    cmsg_outs.try_reserve(msg_count).or(Err(Errno::ENOMEM))?;

    let mmsghdr_size = if is32 {
        size_of::<mmsghdr32>()
    } else {
        size_of::<mmsghdr>()
    };
    let hdr_len = msg_count
        .checked_mul(mmsghdr_size)
        .ok_or(Errno::EOVERFLOW)?;

    let mut hdr_buf: Zeroizing<Vec<u8>> = Zeroizing::new(Vec::new());
    hdr_buf.try_reserve(hdr_len).or(Err(Errno::ENOMEM))?;
    hdr_buf.resize(hdr_len, 0);

    // Allocate address buffer which is one sockaddr_storage per message.
    let addr_buf_len = msg_count
        .checked_mul(SOCKADDR_SIZE)
        .ok_or(Errno::EOVERFLOW)?;
    let mut addr_buf: Zeroizing<Vec<u8>> = Zeroizing::new(Vec::new());
    addr_buf.try_reserve(addr_buf_len).or(Err(Errno::ENOMEM))?;
    addr_buf.resize(addr_buf_len, 0);

    // Allocate per-message address metadata.
    #[expect(clippy::type_complexity)]
    let mut addr_meta: Vec<Option<(u64, usize, usize)>> = Vec::new();
    addr_meta.try_reserve(msg_count).or(Err(Errno::ENOMEM))?;

    // Track blocking call for invalidation semantics.
    let is_blocking = !is_nonblock && !call_flags.contains(MsgFlags::MSG_DONTWAIT);
    let ignore_restart = if is_blocking {
        timeout.is_some() || has_recv_timeout(&fd)?
    } else {
        false
    };
    if is_blocking {
        request.cache.add_sys_block(req, ignore_restart)?;
    }

    // Perform recvmmsg(2).
    let result = recvmmsg(&fd, &mut msgs[..msg_count], call_flags, timeout.as_mut());

    // Remove invalidation record.
    if is_blocking {
        request.cache.del_sys_block(req.id)?;
    }

    // Check result after critical block.
    let msg_count = result?;

    // Iterate over raw mmsghdr results.
    //
    // Linux returns number of processed messages when a later iteration fails.
    for (idx, mmsg_hdr) in msgs.iter_mut().enumerate().take(msg_count) {
        let mmsg_hdr = mmsg_hdr.as_inner_mut();

        // Restore msg_iov pointer.
        if let Some(iov_ptr) = user_iov_bases.get(idx).copied().flatten() {
            mmsg_hdr.msg_hdr.msg_iov = iov_ptr as *mut iovec;
        }

        // Prepare peer address.
        let addr_out = if let Some((nam_ptr, nam_len)) = user_nam_bases.get(idx).copied().flatten()
        {
            let addr_off = idx.checked_mul(SOCKADDR_SIZE).ok_or(Errno::EOVERFLOW)?;
            #[expect(clippy::arithmetic_side_effects)]
            let addr_buf = &mut addr_buf[addr_off..addr_off + SOCKADDR_SIZE];
            let (namelen_out, addr_len) = if sock_dom == AddressFamily::Unix {
                fixup_unix_addr(
                    &fd,
                    request,
                    mmsg_hdr.msg_hdr.msg_name,
                    mmsg_hdr.msg_hdr.msg_namelen,
                    addr_buf,
                )?
            } else {
                copy_addr(
                    mmsg_hdr.msg_hdr.msg_name,
                    mmsg_hdr.msg_hdr.msg_namelen,
                    addr_buf,
                )?
            };

            // Fixup message header pointers.
            mmsg_hdr.msg_hdr.msg_namelen = namelen_out;
            mmsg_hdr.msg_hdr.msg_name = nam_ptr as *mut c_void;

            if addr_len > 0 {
                Some((nam_ptr, nam_len, addr_len))
            } else {
                None
            }
        } else {
            None
        };

        // Prepare control messages.
        #[expect(clippy::disallowed_methods)]
        #[expect(clippy::useless_conversion)]
        let cmsg_out = if let Some((ctl_ptr, ctl_len)) = user_ctl_bases.get(idx).copied().flatten()
        {
            if !mmsg_hdr.msg_hdr.msg_control.is_null() && mmsg_hdr.msg_hdr.msg_controllen > 0 {
                // SAFETY: msg_control and msg_controllen are valid.
                #[expect(clippy::unnecessary_cast)]
                let cmsg_buf = unsafe {
                    std::slice::from_raw_parts(
                        mmsg_hdr.msg_hdr.msg_control as *const u8,
                        mmsg_hdr.msg_hdr.msg_controllen as usize,
                    )
                };
                let close_on_exec =
                    flags.force_cloexec() || call_flags.contains(MsgFlags::MSG_CMSG_CLOEXEC);
                let rand_fd = flags.force_rand_fd();

                let (cmsgs, cmsgs_truncated) =
                    request.fixup_cmsgs(&fd, cmsg_buf, ctl_len, close_on_exec, rand_fd)?;
                let (out_buf, cmsg_len, truncated) = request.setup_cmsgs(&cmsgs, ctl_len)?;
                if truncated || cmsgs_truncated {
                    mmsg_hdr.msg_hdr.msg_flags |= MsgFlags::MSG_CTRUNC.bits();
                }

                mmsg_hdr.msg_hdr.msg_control = ctl_ptr as *mut c_void;
                // unwrap is for musl compat.
                mmsg_hdr.msg_hdr.msg_controllen = cmsg_len.try_into().unwrap();

                Some(out_buf)
            } else {
                mmsg_hdr.msg_hdr.msg_controllen = 0;

                None
            }
        } else {
            mmsg_hdr.msg_hdr.msg_controllen = 0;

            None
        };

        // Copy header into pre-allocated buffer.
        let hdr_off = idx.checked_mul(mmsghdr_size).ok_or(Errno::EOVERFLOW)?;
        #[expect(clippy::arithmetic_side_effects)]
        let dst = &mut hdr_buf[hdr_off..hdr_off + mmsghdr_size];
        if is32 {
            let m32: mmsghdr32 = (*mmsg_hdr).try_into()?;
            let buf: [u8; size_of::<mmsghdr32>()] = m32.to_byte_array();
            dst.copy_from_slice(&buf);
        } else {
            let m64: mmsghdr = (*mmsg_hdr).into();
            let buf: [u8; size_of::<mmsghdr>()] = m64.to_byte_array();
            dst.copy_from_slice(&buf);
        }

        cmsg_outs.push(cmsg_out);
        addr_meta.push(addr_out);
    }

    // Build per-message iovs.
    //
    // For each message push payload, control message, header, and address.
    for idx in 0..msg_count {
        let msg = msgs[idx].as_inner_mut();

        // Copy iov payload scatter entries.
        if let Some(bufs) = msg_bufs.get(idx).and_then(Option::as_ref) {
            scatter_iov(msg.msg_len as usize, bufs, &mut iovs_l, &mut iovs_r)?;
        }

        // Copy control message.
        if let Some(ref out_buf) = cmsg_outs[idx] {
            #[expect(clippy::unnecessary_cast)]
            let cmsg_len = msg.msg_hdr.msg_controllen as usize;
            if cmsg_len > 0 {
                if let Some((ctl_ptr, _)) = user_ctl_bases.get(idx).copied().flatten() {
                    iovs_l.push(IoSlice::new(&out_buf[..cmsg_len]));
                    iovs_r.push(RemoteIoVec {
                        base: usize::try_from(ctl_ptr).or(Err(Errno::EOVERFLOW))?,
                        len: cmsg_len,
                    });
                }
            }
        }

        // Copy header.
        let hdr_off = idx.checked_mul(mmsghdr_size).ok_or(Errno::EOVERFLOW)?;
        let hdr_len = (idx as u64)
            .checked_mul(mmsghdr_size as u64)
            .ok_or(Errno::EOVERFLOW)?;
        let off = msgs_offset.checked_add(hdr_len).ok_or(Errno::EOVERFLOW)?;

        #[expect(clippy::arithmetic_side_effects)]
        iovs_l.push(IoSlice::new(&hdr_buf[hdr_off..hdr_off + mmsghdr_size]));
        iovs_r.push(RemoteIoVec {
            base: usize::try_from(off).or(Err(Errno::EOVERFLOW))?,
            len: mmsghdr_size,
        });

        // Copy peer address.
        if let Some((nam_ptr, nam_len, addr_len)) = addr_meta[idx] {
            let namelen_out = msg.msg_hdr.msg_namelen;

            #[expect(clippy::cast_possible_truncation)]
            let out_len = (namelen_out.min(nam_len as socklen_t)) as usize;
            let out_len = out_len.min(addr_len);

            if out_len > 0 {
                let addr_off = idx.checked_mul(SOCKADDR_SIZE).ok_or(Errno::EOVERFLOW)?;
                #[expect(clippy::arithmetic_side_effects)]
                iovs_l.push(IoSlice::new(&addr_buf[addr_off..addr_off + out_len]));
                iovs_r.push(RemoteIoVec {
                    base: usize::try_from(nam_ptr).or(Err(Errno::EOVERFLOW))?,
                    len: out_len,
                });
            }
        }
    }

    // Copy timeout which is not part of any message.
    let mut timeout_buf: Zeroizing<[u8; size_of::<TimeSpec64>()]> =
        Zeroizing::new([0u8; size_of::<TimeSpec64>()]);
    let timeout_len: usize = if msg_count > 0 {
        if let Some(timeout) = timeout {
            if timeout_is32 {
                let t32: TimeSpec32 = timeout.try_into()?;
                let buf: [u8; size_of::<TimeSpec32>()] = t32.to_byte_array();
                timeout_buf[..buf.len()].copy_from_slice(&buf);
                buf.len()
            } else {
                let buf: [u8; size_of::<TimeSpec64>()] = timeout.to_byte_array();
                timeout_buf.copy_from_slice(&buf);
                buf.len()
            }
        } else {
            0
        }
    } else {
        0
    };

    // Copy timeout as necessary.
    if timeout_len > 0 {
        iovs_l.push(IoSlice::new(&timeout_buf[..timeout_len]));
        iovs_r.push(RemoteIoVec {
            base: usize::try_from(args[4]).or(Err(Errno::EOVERFLOW))?,
            len: timeout_len,
        });
    }

    // Write in a single batch.
    //
    // Linux returns count of messages on partial writes.
    let mut msg_count = msg_count;
    if !iovs_l.is_empty() {
        let len = request.write_mem_many_all(&iovs_l, &iovs_r)?;
        let siz: usize = iovs_r.iter().map(|v| v.len).sum();

        // Detect partial writes an return count of messages.
        if len != siz {
            let mut off = 0usize;
            let mut nbytes = 0usize;
            let mut nwrite = 0usize;

            #[expect(clippy::arithmetic_side_effects)]
            for idx in 0..msg_count {
                let n = msg_iov_count(
                    &msgs[idx],
                    msg_bufs.get(idx).and_then(Option::as_ref),
                    &cmsg_outs[idx],
                    user_ctl_bases.get(idx).copied().flatten(),
                    &addr_meta[idx],
                );

                nbytes += iovs_r[off..off + n].iter().map(|v| v.len).sum::<usize>();
                if nbytes > len {
                    break;
                }

                nwrite += 1;
                off += n;
            }

            if nwrite == 0 {
                return Err(Errno::EFAULT);
            }

            msg_count = nwrite;
        }
    }

    // Return number of messages received.
    #[expect(clippy::cast_possible_wrap)]
    Ok(request.return_syscall(msg_count as i64))
}

#[expect(clippy::type_complexity)]
fn msg_iov_count(
    mmsg: &MmsgHdr,
    bufs: Option<&Vec<(Zeroizing<Vec<u8>>, u64)>>,
    cmsg_out: &Option<Zeroizing<Vec<u8>>>,
    ctl_base: Option<(u64, usize)>,
    addr_meta: &Option<(u64, usize, usize)>,
) -> usize {
    let mut n = 0usize;

    // Count payload:
    // One iov per scatter buffer entry up to msg_len bytes.
    if let Some(bufs) = bufs {
        let mut nrem = mmsg.msg_len() as usize;
        if nrem > 0 {
            #[expect(clippy::arithmetic_side_effects)]
            for (buf, _) in bufs {
                if nrem == 0 {
                    break;
                }
                n += 1;
                nrem = nrem.saturating_sub(buf.len());
            }
        }
    }

    // Count control message if present.
    if cmsg_out.is_some() {
        #[expect(clippy::unnecessary_cast)]
        let cmsg_len = mmsg.as_inner().msg_hdr.msg_controllen as usize;
        #[expect(clippy::arithmetic_side_effects)]
        if cmsg_len > 0 && ctl_base.is_some() {
            n += 1;
        }
    }

    // Count message header.
    #[expect(clippy::arithmetic_side_effects)]
    {
        n += 1;
    }

    // Count address if present.
    if let Some((_, nam_len, addr_len)) = addr_meta {
        let namelen_out = mmsg.as_inner().msg_hdr.msg_namelen;
        #[expect(clippy::cast_possible_truncation)]
        let out_len = (namelen_out.min(*nam_len as socklen_t)) as usize;
        let out_len = out_len.min(*addr_len);
        #[expect(clippy::arithmetic_side_effects)]
        if out_len > 0 {
            n += 1;
        }
    }

    n
}

// Gather iov payload scatter entries into pre-allocated buffers.
#[expect(clippy::type_complexity)]
fn scatter_iov<'a>(
    size: usize,
    bufs: &'a [(Zeroizing<Vec<u8>>, u64)],
    local_iovs: &mut Vec<IoSlice<'a>>,
    remote_iovs: &mut Vec<RemoteIoVec>,
) -> Result<(), Errno> {
    if size == 0 || bufs.is_empty() {
        return Ok(());
    }

    let mut nrem = size;
    for (buf, ptr) in bufs {
        if nrem == 0 {
            break;
        }
        let take = nrem.min(buf.len());

        local_iovs.push(IoSlice::new(&buf[..take]));
        remote_iovs.push(RemoteIoVec {
            base: usize::try_from(*ptr).or(Err(Errno::EFAULT))?,
            len: take,
        });

        nrem = nrem.checked_sub(take).ok_or(Errno::EOVERFLOW)?;
    }

    Ok(())
}

// Resolve Unix peer address from a kernel-returned message header,
// and copy into given buffer.
//
// Returns new "msg_namelen" value and address bytes to copy.
fn fixup_unix_addr<Fd: AsFd>(
    fd: Fd,
    request: &UNotifyEventRequest,
    msg_name: *mut c_void,
    msg_namelen: socklen_t,
    addr_buf: &mut [u8],
) -> Result<(socklen_t, usize), Errno> {
    let r_addr = if !msg_name.is_null() && msg_namelen > 0 {
        // SAFETY: msg_name and msg_namelen are from a kernel-returned msghdr.
        unsafe { SockaddrStorage::from_raw(msg_name as *const sockaddr, Some(msg_namelen)) }
    } else {
        None
    };

    if let Some(mut addr) = r_addr {
        let hdr_namelen = if let Ok(ino) = fd_inode(fd) {
            if let Ok(peer_addr) = request.resolve_unix_peer(&addr, ino) {
                addr = peer_addr;
                addr.as_unix_addr().map_or(addr.len(), unix_addr_len)
            } else {
                msg_namelen
            }
        } else {
            msg_namelen
        };

        // SAFETY:
        // 1. SockaddrStorage is initialized.
        // 2. as_ptr() and len() return valid bounds.
        let src =
            unsafe { std::slice::from_raw_parts(addr.as_ptr().cast::<u8>(), addr.len() as usize) };

        let namelen_out = addr.len().min(hdr_namelen);
        let len = src.len().min(addr_buf.len());

        addr_buf[..len].copy_from_slice(&src[..len]);

        Ok((namelen_out, len))
    } else {
        Ok((0, 0))
    }
}

// Copy address bytes from a kernel-returned message header into given buffer.
//
// Returns new "msg_namelen" value and address bytes to copy.
fn copy_addr(
    msg_name: *mut c_void,
    msg_namelen: socklen_t,
    addr_buf: &mut [u8],
) -> Result<(socklen_t, usize), Errno> {
    if msg_name.is_null() || msg_namelen == 0 {
        return Ok((0, 0));
    }

    let len = msg_namelen as usize;
    if len > addr_buf.len() {
        return Err(Errno::EINVAL);
    }

    // SAFETY: msg_name and msg_namelen are from a kernel-returned msghdr.
    let src = unsafe { std::slice::from_raw_parts(msg_name as *const u8, len) };
    addr_buf[..len].copy_from_slice(src);

    Ok((msg_namelen, len))
}