netring 0.29.0

//! AF_XDP backend for kernel-bypass packet I/O.
//!
//! Provides high-throughput packet I/O via XDP sockets with shared UMEM memory.
//! Uses direct `libc` syscalls (same pure Rust approach as AF_PACKET — no C
//! library dependencies beyond libc).
//!
//! # Ring model
//!
//! AF_XDP uses 4 shared rings over a UMEM region:
//! - **Fill ring**: userspace gives empty frame addrs to kernel for RX
//! - **RX ring**: kernel delivers received packet descriptors to userspace
//! - **TX ring**: userspace submits packet descriptors for transmission
//! - **Completion ring**: kernel returns transmitted frame addrs to userspace
//!
//! # Requirements
//!
//! - Linux 5.4+
//! - `CAP_NET_RAW` (or root) for socket creation
//! - XDP-capable NIC driver for zero-copy mode
//! - An external XDP BPF program (e.g. via `aya`) for RX — not needed for TX-only
//!
//! # Standalone API
//!
//! AF_XDP uses different ring semantics than AF_PACKET (block-based mmap).
//! This module provides a standalone API that does **not** implement
//! [`PacketSource`](crate::traits::PacketSource). A unified trait-based API
//! is planned for a future version using GATs.
//!
//! # Feature gate
//!
//! Requires the `af-xdp` feature. Without it, only the builder types are
//! available (for downstream crates to compile against).

#[cfg(feature = "af-xdp")]
mod batch;
#[cfg(feature = "af-xdp")]
mod capture;
pub(crate) mod ffi;
#[cfg(feature = "af-xdp")]
mod metadata;
#[cfg(feature = "af-xdp")]
mod ring;
#[cfg(feature = "af-xdp")]
pub mod rss;
#[cfg(feature = "af-xdp")]
mod socket;
mod stats;
#[cfg(feature = "af-xdp")]
pub mod steer;
#[cfg(feature = "af-xdp")]
mod umem;

#[cfg(feature = "xdp-loader")]
pub mod loader;

#[cfg(feature = "af-xdp")]
pub use batch::{XdpBatch, XdpBatchIter, XdpPacket};
#[cfg(feature = "af-xdp")]
pub use capture::{Queues, interface_numa_node, queue_count};
#[cfg(all(feature = "af-xdp", feature = "xdp-loader"))]
pub use capture::{XdpCapture, XdpCaptureBuilder, XdpCaptureGuard};
pub use stats::XdpStats;

#[cfg(feature = "xdp-loader")]
pub use loader::{XdpAttachment, XdpFlags, XdpProgram, default_program};

use std::os::fd::{AsFd, AsRawFd};
#[cfg(feature = "af-xdp")]
use std::os::fd::{BorrowedFd, OwnedFd};
#[cfg(feature = "af-xdp")]
use std::time::Duration;

#[cfg(feature = "af-xdp")]
use ring::{CompletionRing, FillRing, RxRing, TxRing};
#[cfg(feature = "af-xdp")]
use umem::Umem;

use crate::error::Error;
#[cfg(feature = "af-xdp")]
use crate::packet::OwnedPacket;

// ── XdpMode ──────────────────────────────────────────────────────────────

/// Operating mode for an AF_XDP socket.
///
/// Controls how UMEM frames are partitioned between the fill ring (for kernel
/// RX) and the free list (available for TX allocation). The wrong split breaks
/// either RX (kernel can't enqueue packets — `rx_dropped` counter rises) or TX
/// (`send()` can never allocate a frame).
///
/// Pick the mode that matches your traffic direction. For asymmetric splits or
/// shared-UMEM setups, use [`XdpMode::Custom`].
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Default)]
#[non_exhaustive]
pub enum XdpMode {
    /// Receive only. All UMEM frames are pre-staged to the fill ring; `send()`
    /// will always return `Ok(false)` because no frames are reachable from the
    /// free list.
    Rx,
    /// Transmit only. No prefill — every UMEM frame stays in the free list and
    /// is available for `send()` allocation. RX descriptors will not arrive
    /// (the fill ring is empty) but stats are unaffected.
    Tx,
    /// Bidirectional. Half the frames prefilled to the fill ring (RX), half
    /// retained in the free list (TX). For uneven traffic patterns prefer
    /// [`XdpMode::Custom`] to control the split explicitly.
    #[default]
    RxTx,
    /// Custom prefill: pre-stage exactly `prefill` frames in the fill ring; the
    /// remaining `frame_count - prefill` stay in the free list for TX.
    ///
    /// `prefill` is clamped to `min(frame_count, ring_size)`. Use `0` to skip
    /// prefill entirely (equivalent to [`XdpMode::Tx`]); use `frame_count` to
    /// prefill everything (equivalent to [`XdpMode::Rx`]).
    Custom {
        /// Number of frames to pre-stage in the fill ring at construction.
        prefill: u32,
    },
}

// ── XdpSocketBuilder ─────────────────────────────────────────────────────

/// Builder for AF_XDP sockets.
///
/// For TX-only workloads, set [`XdpMode::Tx`] — by default the builder
/// pre-stages half the UMEM into the fill ring (RxTx mode), which leaves only
/// half available for `send()`.
///
/// # Examples
///
/// ```no_run,ignore
/// use netring::afxdp::{XdpSocketBuilder, XdpMode};
///
/// let mut xdp = XdpSocketBuilder::default()
///     .interface("eth0")
///     .queue_id(0)
///     .mode(XdpMode::Tx)  // skip RX prefill so send() can allocate frames
///     .build()
///     .unwrap();
///
/// xdp.send(&[0xff; 64]).unwrap();
/// xdp.flush().unwrap();
/// ```
#[derive(Debug)]
#[must_use]
pub struct XdpSocketBuilder {
    interface: Option<String>,
    queue_id: u32,
    frame_size: usize,
    frame_count: usize,
    need_wakeup: bool,
    mode: XdpMode,
    /// Raw fd of an already-bound XDP socket whose UMEM we want to share.
    /// `0` means "no sharing". When non-zero, build() skips
    /// `XDP_UMEM_REG` and sets `XDP_SHARED_UMEM` in the bind flags.
    shared_umem_fd: u32,
    busy_poll_us: Option<u32>,
    prefer_busy_poll: Option<bool>,
    busy_poll_budget: Option<u16>,
    /// 0.25 W4: back the UMEM with hugepages (`MAP_HUGETLB`).
    hugepages: bool,
    /// 0.25 W4: bind the UMEM to this NUMA node (`mbind`).
    numa_node: Option<u32>,
    /// Put the interface into promiscuous mode for the socket's lifetime,
    /// via an auxiliary AF_PACKET `PACKET_MR_PROMISC` guard (issue #4).
    promiscuous: bool,
    /// Reserve UMEM headroom and read the XDP RX-metadata struct ahead of each
    /// frame (issue #13). Requires a matching `redirect_meta` program; degrades
    /// cleanly to no metadata otherwise.
    rx_metadata: bool,
    #[cfg(feature = "xdp-loader")]
    attach_default: bool,
    #[cfg(feature = "xdp-loader")]
    attach_flags: loader::XdpFlags,
    #[cfg(feature = "xdp-loader")]
    force_replace: bool,
    /// Caller-supplied XDP program to register + attach during `build()`.
    /// Mutually exclusive with `attach_default`.
    #[cfg(feature = "xdp-loader")]
    program: Option<loader::XdpProgram>,
}

impl Default for XdpSocketBuilder {
    fn default() -> Self {
        Self {
            interface: None,
            queue_id: 0,
            frame_size: 4096,
            frame_count: 4096,
            need_wakeup: true,
            mode: XdpMode::default(),
            shared_umem_fd: 0,
            busy_poll_us: None,
            prefer_busy_poll: None,
            busy_poll_budget: None,
            hugepages: false,
            numa_node: None,
            promiscuous: false,
            rx_metadata: false,
            #[cfg(feature = "xdp-loader")]
            attach_default: false,
            #[cfg(feature = "xdp-loader")]
            attach_flags: loader::XdpFlags::SKB_MODE, // safest default; works on lo
            #[cfg(feature = "xdp-loader")]
            force_replace: false,
            #[cfg(feature = "xdp-loader")]
            program: None,
        }
    }
}

impl XdpSocketBuilder {
    /// Set the network interface name (required).
    pub fn interface(mut self, name: &str) -> Self {
        self.interface = Some(name.to_string());
        self
    }

    /// Set the NIC queue ID to bind to. Default: 0.
    pub fn queue_id(mut self, id: u32) -> Self {
        self.queue_id = id;
        self
    }

    /// UMEM frame size. Default: 4096.
    pub fn frame_size(mut self, size: usize) -> Self {
        self.frame_size = size;
        self
    }

    /// Number of UMEM frames. Default: 4096.
    pub fn frame_count(mut self, count: usize) -> Self {
        self.frame_count = count;
        self
    }

    /// 0.25 W4: back the UMEM with **hugepages** (`MAP_HUGETLB`, 2 MiB) to cut
    /// TLB misses on the per-frame DMA path. Default: false.
    ///
    /// Requires hugepages reserved on the host
    /// (`sysctl vm.nr_hugepages` / `hugeadm`). Best-effort: if the hugepage
    /// mapping fails (none reserved), `build()` falls back to regular pages with
    /// a `warn` rather than failing. The UMEM region is rounded up to a
    /// hugepage multiple.
    pub fn hugepages(mut self, enable: bool) -> Self {
        self.hugepages = enable;
        self
    }

    /// 0.25 W4: bind the UMEM's pages to NUMA `node` (`mbind` / `MPOL_BIND`) —
    /// set this to the NIC's NUMA node to avoid cross-node DMA + cache traffic.
    /// Best-effort: a binding failure (single-node host, missing privilege) is
    /// logged at `warn` and ignored.
    pub fn numa_node(mut self, node: u32) -> Self {
        self.numa_node = Some(node);
        self
    }

    /// Put the interface into **promiscuous mode** for the lifetime of the
    /// returned [`XdpSocket`]. Default: `false`.
    ///
    /// AF_XDP runs in the driver RX path *after* the NIC's MAC filter, so on a
    /// non-promiscuous interface the socket only sees frames addressed to that
    /// NIC (plus broadcast and subscribed multicast). Enable this to capture
    /// all traffic on the wire — the usual need for SPAN/mirror ports, passive
    /// sniffing, or bridging.
    ///
    /// Promiscuity is a `netdev` property, not an AF_XDP socket option, so
    /// netring holds it through an auxiliary AF_PACKET socket joined to
    /// `PACKET_MR_PROMISC` — the same mechanism the AF_PACKET capture path
    /// uses. The kernel reference-counts `dev->promiscuity` and releases it
    /// automatically when the `XdpSocket` is dropped, including on crash (the
    /// kernel closes fds on process exit). No manual restore, no leak.
    ///
    /// Requires `CAP_NET_RAW`, which AF_XDP capture already needs. Two gotchas
    /// worth knowing:
    /// - `PACKET_MR_PROMISC` does **not** set the user-visible `IFF_PROMISC`
    ///   flag, so `ip link` / `ifconfig` will not display `PROMISC` even though
    ///   the interface *is* in promiscuous mode (`dev->promiscuity` is raised).
    /// - On a multi-queue NIC an XSK binds to a single queue, and RSS spreads
    ///   traffic across queues, so one socket still only sees its queue's share
    ///   even in promiscuous mode. For complete capture, force a single queue
    ///   (`ethtool -L <iface> combined 1`) or open one socket per queue.
    pub fn promiscuous(mut self, enable: bool) -> Self {
        self.promiscuous = enable;
        self
    }

    /// Read NIC hardware RX metadata — timestamp, RSS hash, VLAN tag — per
    /// frame (issue #13).
    ///
    /// Reserves UMEM headroom for the metadata struct the `redirect_meta` XDP
    /// program writes ahead of each frame (kernel 6.3+, via the
    /// `bpf_xdp_metadata_*` kfuncs); the parsed values surface through
    /// [`XdpPacket::rx_metadata`](crate::XdpPacket::rx_metadata) and as the
    /// packet [`timestamp`](crate::XdpPacket::timestamp).
    ///
    /// Hardware-gated: drivers without the kfuncs (and the loopback / generic
    /// XDP path) return nothing, and metadata degrades to absent with the
    /// software timestamp used as before. Off by default.
    pub fn rx_metadata(mut self, enable: bool) -> Self {
        self.rx_metadata = enable;
        self
    }

    /// Operating mode (RX/TX/RxTx/Custom prefill split). Default: [`XdpMode::RxTx`].
    ///
    /// For TX-only workloads — most importantly the [`xdp_send`] example pattern —
    /// you **must** set this to [`XdpMode::Tx`], or every `send()` call will
    /// silently fail because all UMEM frames are pre-staged to the fill ring.
    ///
    /// [`xdp_send`]: https://github.com/p13marc/netring/blob/master/examples/xdp_send.rs
    pub fn mode(mut self, mode: XdpMode) -> Self {
        self.mode = mode;
        self
    }

    /// Enable `XDP_USE_NEED_WAKEUP` optimization. Default: true.
    pub fn need_wakeup(mut self, enable: bool) -> Self {
        self.need_wakeup = enable;
        self
    }

    /// Enable `SO_BUSY_POLL` with the given microsecond timeout. Kernel ≥ 4.5.
    ///
    /// For low-latency AF_XDP capture, pair with [`prefer_busy_poll`](Self::prefer_busy_poll)
    /// and [`busy_poll_budget`](Self::busy_poll_budget). See
    /// <https://docs.kernel.org/networking/af_xdp.html>.
    pub fn busy_poll_us(mut self, us: u32) -> Self {
        self.busy_poll_us = Some(us);
        self
    }

    /// Enable `SO_PREFER_BUSY_POLL`. Kernel ≥ 5.11.
    ///
    /// Tells the kernel to prefer the busy-polling path over softirq for
    /// this socket. Has no effect without [`busy_poll_us`](Self::busy_poll_us)
    /// also set.
    pub fn prefer_busy_poll(mut self, enable: bool) -> Self {
        self.prefer_busy_poll = Some(enable);
        self
    }

    /// Set `SO_BUSY_POLL_BUDGET` (per-poll packet cap). Kernel ≥ 5.11.
    ///
    /// Default kernel budget is 8 in 6.x; 64 is a common production value
    /// for AF_XDP. Values above `/proc/sys/net/core/busy_poll_budget_max`
    /// (typically 64) require `CAP_NET_ADMIN`; otherwise the kernel returns
    /// `EPERM`.
    pub fn busy_poll_budget(mut self, budget: u16) -> Self {
        self.busy_poll_budget = Some(budget);
        self
    }

    /// Load and attach the built-in redirect-all XDP program, then
    /// register this socket on its embedded XSKMAP. Available with
    /// the `xdp-loader` feature.
    ///
    /// On `Drop` of the resulting `XdpSocket`, the program is
    /// detached from the interface and the map fd is closed.
    ///
    /// Default attach mode is `SKB_MODE` (works on every interface
    /// including `lo`). Override with [`xdp_attach_flags`](Self::xdp_attach_flags)
    /// for `DRV_MODE` on supported NICs.
    ///
    /// If a program is already attached to the interface, `build()`
    /// fails with an error suggesting [`force_replace`](Self::force_replace).
    #[cfg(feature = "xdp-loader")]
    pub fn with_default_program(mut self) -> Self {
        self.attach_default = true;
        self
    }

    /// Use a caller-loaded XDP program. Like
    /// [`with_default_program`](Self::with_default_program) but for
    /// programs you've compiled and loaded yourself (e.g. via
    /// `aya-bpf` + `bpf-linker`). The program must contain a
    /// `BPF_MAP_TYPE_XSKMAP` and call
    /// `bpf_redirect_map(&xsks_map, ctx->rx_queue_index, ...)`.
    ///
    /// On `build()`, netring registers this socket on the program's
    /// XSKMAP at the configured `queue_id`, then attaches the program
    /// to the interface using `xdp_attach_flags` /
    /// `force_replace`. The attachment lives as long as the
    /// returned `XdpSocket` (RAII detach on drop).
    ///
    /// Mutually exclusive with [`with_default_program`](Self::with_default_program);
    /// `build()` returns an error if both are set.
    ///
    /// For the multi-queue case (one program serving many sockets),
    /// don't use this method — instead manage [`XdpProgram::attach`]
    /// and [`XdpProgram::register`] manually so the attachment lives
    /// across socket lifetimes.
    ///
    /// ```no_run
    /// # #[cfg(all(feature = "af-xdp", feature = "xdp-loader"))]
    /// # fn main() -> Result<(), netring::Error> {
    /// use aya::Ebpf;
    /// use netring::xdp::{XdpFlags, XdpProgram};
    /// use netring::XdpSocket;
    ///
    /// let bytecode: &[u8] = b""; // your compiled program
    /// let bpf = Ebpf::load(bytecode).unwrap();
    /// let prog = XdpProgram::from_aya(bpf, "xdp_sock_prog", "xsks_map");
    ///
    /// let xsk = XdpSocket::builder()
    ///     .interface("eth0")
    ///     .queue_id(0)
    ///     .with_program(prog)
    ///     .xdp_attach_flags(XdpFlags::DRV_MODE)
    ///     .build()?;
    /// # Ok(()) }
    /// # #[cfg(not(all(feature = "af-xdp", feature = "xdp-loader")))]
    /// # fn main() {}
    /// ```
    #[cfg(feature = "xdp-loader")]
    pub fn with_program(mut self, prog: loader::XdpProgram) -> Self {
        self.program = Some(prog);
        self
    }

    /// Set the XDP attach flags used by [`with_default_program`](Self::with_default_program).
    /// Default: `SKB_MODE`.
    #[cfg(feature = "xdp-loader")]
    pub fn xdp_attach_flags(mut self, flags: loader::XdpFlags) -> Self {
        self.attach_flags = flags;
        self
    }

    /// Request replacement of an existing XDP program on the interface.
    ///
    /// **Kernel ≥ 5.9 caveat:** netring (via aya) attaches through the
    /// `bpf_link_create` link API, which does **not** support
    /// `XDP_FLAGS_REPLACE` — that flag is only honoured by the legacy netlink
    /// attach path. So on modern kernels atomic replacement of a *foreign*
    /// incumbent program is not reachable here. Setting this flag therefore no
    /// longer forces a replace; instead it makes a failed attach return an
    /// actionable [`Error::Loader`] explaining how to clear the incumbent
    /// (`ip link set dev <iface> xdp off`). (Previously this OR'd in
    /// `XDP_FLAGS_REPLACE` and crashed with `EINVAL`.)
    ///
    /// Default: false (build fails with the raw attach error if a program is
    /// already attached).
    #[cfg(feature = "xdp-loader")]
    pub fn force_replace(mut self, force: bool) -> Self {
        self.force_replace = force;
        self
    }

    /// Bind this socket as a secondary that **shares the UMEM** of an
    /// existing AF_XDP socket.
    ///
    /// Pass an `AsFd` for an already-bound `XdpSocket` that owns the
    /// UMEM you want to reuse. The kernel skips per-socket UMEM
    /// registration and threads packets through *this* socket's RX/TX
    /// rings while addressing the same shared frame pool.
    ///
    /// # Manual partitioning
    ///
    /// netring does **not** synchronize allocation between the primary
    /// and secondary sockets. Each socket's allocator hands out frame
    /// addresses independently — if both pick the same address, the
    /// kernel will overwrite one with the other's data. You're
    /// responsible for partitioning the UMEM range across sockets.
    /// The simplest scheme is: primary uses the first half, secondary
    /// uses the second half; configure each via `frame_count` and
    /// arrange for them to land at disjoint frame indices.
    ///
    /// # Expert-only — prefer per-socket UMEM for multi-queue capture
    ///
    /// Issue #6 F1: the high-level [`XdpCapture`] gives
    /// each per-queue socket its **own** UMEM and is the blessed multi-queue
    /// path. Shared UMEM is a *memory* optimization with two sharp edges, so it
    /// stays low-level and opt-in:
    /// 1. **Manual frame-space partitioning** (above) — netring won't do it for
    ///    you, and a mistake silently corrupts frames.
    /// 2. **A per-CPU FILL-queue race** — when sockets on different cores share
    ///    one UMEM, the kernel's shared `xsk_buff_pool` FILL queue can race
    ///    ([kernel AF_XDP docs] / LKML 2025). Only share a UMEM across sockets
    ///    you drive from a **single thread**, or accept the partitioning + FILL
    ///    discipline yourself.
    ///
    /// Reach for this only to cut UMEM memory on a single-threaded multi-queue
    /// drain; otherwise use `XdpCapture` (own UMEM per queue).
    ///
    /// [kernel AF_XDP docs]: https://docs.kernel.org/networking/af_xdp.html
    pub fn shared_umem<F: AsFd>(mut self, primary: F) -> Self {
        self.shared_umem_fd = primary.as_fd().as_raw_fd() as u32;
        self
    }

    /// Validate the builder configuration.
    ///
    /// Returns the interface name if valid.
    pub fn validate(&self) -> Result<&str, Error> {
        let iface = self
            .interface
            .as_deref()
            .ok_or_else(|| Error::Config("interface is required".into()))?;
        if self.frame_size == 0 {
            return Err(Error::Config("frame_size must be > 0".into()));
        }
        if self.frame_count == 0 {
            return Err(Error::Config("frame_count must be > 0".into()));
        }
        Ok(iface)
    }

    /// Build the XDP socket.
    ///
    /// # Errors
    ///
    /// - [`Error::Config`] if configuration is invalid
    /// - [`Error::PermissionDenied`] if missing `CAP_NET_RAW`
    /// - [`Error::Socket`], [`Error::SockOpt`], [`Error::Mmap`], [`Error::Bind`]
    ///   for the respective syscall failures
    #[cfg(feature = "af-xdp")]
    // `self` is only mutated on the `xdp-loader` path (program attach); without
    // that feature the binding below is genuinely immutable.
    #[cfg_attr(not(feature = "xdp-loader"), allow(unused_mut))]
    pub fn build(mut self) -> Result<XdpSocket, Error> {
        let iface = self.validate()?;
        let ifindex = crate::afpacket::socket::resolve_interface(iface)? as u32;

        // 1. Allocate UMEM (MAP_PRIVATE | MAP_ANONYMOUS | MAP_POPULATE)
        let mut umem = Umem::new_with_options(
            self.frame_size,
            self.frame_count,
            &umem::UmemOptions {
                hugepages: self.hugepages,
                numa_node: self.numa_node,
                headroom: if self.rx_metadata {
                    metadata::XdpRxMeta::LEN as u32
                } else {
                    0
                },
            },
        )?;

        // 2. Create AF_XDP socket
        let fd = socket::create_xdp_socket()?;

        // 2a. Apply busy-poll trio if requested. These are SOL_SOCKET options
        //     and work identically on AF_XDP and AF_PACKET sockets, so we reuse
        //     the AF_PACKET helpers.
        if let Some(us) = self.busy_poll_us {
            crate::afpacket::socket::set_busy_poll(fd.as_fd(), us)?;
        }
        if let Some(prefer) = self.prefer_busy_poll {
            crate::afpacket::socket::set_prefer_busy_poll(fd.as_fd(), prefer)?;
        }
        if let Some(budget) = self.busy_poll_budget {
            crate::afpacket::socket::set_busy_poll_budget(fd.as_fd(), budget)?;
        }

        // 3. Register UMEM with kernel.
        //    Skipped when this socket is binding as a XDP_SHARED_UMEM secondary —
        //    the kernel inherits the UMEM from the primary socket fd.
        if self.shared_umem_fd == 0 {
            socket::register_umem(fd.as_fd(), &umem.as_reg())?;
        }

        // 4. Configure ring sizes (each power of 2, independent)
        let ring_size = (self.frame_count as u32).next_power_of_two();
        socket::set_ring_size(
            fd.as_fd(),
            ffi::XDP_UMEM_FILL_RING,
            ring_size,
            "XDP_UMEM_FILL_RING",
        )?;
        socket::set_ring_size(
            fd.as_fd(),
            ffi::XDP_UMEM_COMPLETION_RING,
            ring_size,
            "XDP_UMEM_COMPLETION_RING",
        )?;
        socket::set_ring_size(fd.as_fd(), ffi::XDP_RX_RING, ring_size, "XDP_RX_RING")?;
        socket::set_ring_size(fd.as_fd(), ffi::XDP_TX_RING, ring_size, "XDP_TX_RING")?;

        // 5. Get mmap offsets from kernel
        let offsets = socket::get_mmap_offsets(fd.as_fd())?;

        // 6. mmap all 4 rings
        // NOTE: offsets.fr = fill ring, offsets.cr = completion ring (NOT .fill/.completion)
        let mut fill = unsafe {
            FillRing::mmap(
                fd.as_fd(),
                ring_size,
                &offsets.fr,
                ffi::XDP_UMEM_PGOFF_FILL_RING as libc::off_t,
            )?
        };
        let comp = unsafe {
            CompletionRing::mmap(
                fd.as_fd(),
                ring_size,
                &offsets.cr,
                ffi::XDP_UMEM_PGOFF_COMPLETION_RING as libc::off_t,
            )?
        };
        let rx = unsafe {
            RxRing::mmap(
                fd.as_fd(),
                ring_size,
                &offsets.rx,
                ffi::XDP_PGOFF_RX_RING as libc::off_t,
            )?
        };
        let tx = unsafe {
            TxRing::mmap(
                fd.as_fd(),
                ring_size,
                &offsets.tx,
                ffi::XDP_PGOFF_TX_RING as libc::off_t,
            )?
        };

        // 7. Pre-fill the fill ring with frame addrs for kernel RX. The amount
        //    to prefill depends on the operating mode:
        //      - Rx: every frame goes to the fill ring (TX disabled).
        //      - Tx: nothing prefilled (free list keeps all frames for TX).
        //      - RxTx: half of frames prefilled, half retained for TX.
        //      - Custom: caller-specified count, clamped to [0, min(avail, ring)].
        let cap_avail = umem.available().min(ring_size as usize);
        let prefill = match self.mode {
            XdpMode::Rx => cap_avail,
            XdpMode::Tx => 0,
            XdpMode::RxTx => cap_avail / 2,
            XdpMode::Custom { prefill } => (prefill as usize).min(cap_avail),
        } as u32;

        if prefill > 0
            && let Some(tok) = fill.producer_reserve(prefill)
        {
            let mut written = 0u32;
            for i in 0..prefill {
                match umem.alloc_frame() {
                    Some(addr) => {
                        fill.write_at(tok, i, addr);
                        written += 1;
                    }
                    None => break,
                }
            }
            // Even if we wrote fewer than reserved, the producer index
            // was advanced by `n` in reserve — submit the original token.
            // Unwritten descriptors carry stale data, which is fine as
            // long as the kernel reads only `written` of them. In our
            // case alloc_frame can't fail mid-loop because we capped
            // prefill to umem.available() up front.
            debug_assert_eq!(written, tok.n);
            if written > 0 {
                fill.producer_submit(tok);
            }
        }

        // 8. Bind to interface + queue
        // flags=0: auto-negotiate (kernel tries zero-copy, falls back to copy)
        // XDP_USE_NEED_WAKEUP enables wakeup optimization
        // XDP_SHARED_UMEM tells the kernel to attach this socket to the
        //   UMEM owned by `shared_umem_fd` instead of the one we registered.
        let mut bind_flags: u16 = 0;
        if self.need_wakeup {
            bind_flags |= ffi::XDP_USE_NEED_WAKEUP;
        }
        if self.shared_umem_fd != 0 {
            bind_flags |= ffi::XDP_SHARED_UMEM;
        }
        socket::bind_xdp(
            fd.as_fd(),
            ifindex,
            self.queue_id,
            bind_flags,
            self.shared_umem_fd,
        )?;

        // 0.25 W4 / issue #6 F2: read the bind mode. On a fast NIC the per-frame
        // copy of COPY mode dominates; the operator should know to switch to a
        // native-XDP driver + DRV_MODE for true zero-copy. We both `warn!` and
        // store the result so callers can query it via `is_zerocopy()` instead
        // of grepping logs. Best-effort: a failed read is treated as not-ZC.
        let zerocopy = {
            let mut opts: u32 = 0;
            let mut len = std::mem::size_of::<u32>() as libc::socklen_t;
            // SAFETY: getsockopt writes a u32 into `opts`; `len` matches.
            let rc = unsafe {
                libc::getsockopt(
                    fd.as_raw_fd(),
                    libc::SOL_XDP,
                    libc::XDP_OPTIONS,
                    &mut opts as *mut u32 as *mut libc::c_void,
                    &mut len,
                )
            };
            let zc = rc == 0 && opts & libc::XDP_OPTIONS_ZEROCOPY != 0;
            if self.shared_umem_fd == 0 && !zc {
                tracing::warn!(
                    "AF_XDP bound in COPY mode (not zero-copy); for line-rate use a \
                     native-XDP-capable driver with XdpFlags::DRV_MODE"
                );
            }
            zc
        };

        // Optionally hold the interface in promiscuous mode for the socket's
        // lifetime (issue #4). AF_XDP has no promisc knob; an AF_PACKET
        // `PACKET_MR_PROMISC` guard provides it and self-cleans on drop.
        let promisc_guard = if self.promiscuous {
            Some(crate::afpacket::socket::PromiscGuard::enable(
                ifindex as i32,
            )?)
        } else {
            None
        };

        #[cfg_attr(not(feature = "xdp-loader"), allow(unused_mut))]
        let mut socket = XdpSocket {
            fd,
            umem,
            fill,
            rx,
            tx,
            comp,
            need_wakeup_enabled: self.need_wakeup,
            zerocopy,
            rx_metadata: self.rx_metadata,
            _promisc_guard: promisc_guard,
            #[cfg(feature = "xdp-loader")]
            _xdp_attachment: None,
        };

        // Optionally load + attach an XDP program (built-in or
        // caller-supplied) and register this socket on its XSKMAP.
        #[cfg(feature = "xdp-loader")]
        {
            loader::check_program_conflict(self.attach_default, self.program.is_some())?;
            let prog: Option<loader::XdpProgram> = if let Some(p) = self.program.take() {
                Some(p)
            } else if self.attach_default {
                Some(loader::default_program(/* max_queues */ 256)?)
            } else {
                None
            };
            if let Some(mut prog) = prog {
                let iface = self.interface.as_deref().unwrap_or("");
                // NOTE: `XDP_FLAGS_REPLACE` is **not** OR'd in here even when
                // `force_replace` is set. On kernels ≥ 5.9 aya attaches via
                // `bpf_link_create`, which rejects `XDP_FLAGS_REPLACE` (that flag
                // is only honoured by the legacy netlink `IFLA_XDP_FLAGS` path).
                // Passing it caused `EINVAL` ("bpf_link_create failed"). Atomic
                // replacement of a *foreign* incumbent program isn't reachable
                // through the link API, so `force_replace` now only enriches the
                // attach error with a remediation hint (see below).
                let flags = self.attach_flags;
                // Register *before* attach: the kernel program reads
                // map[queue_id] on the very first packet, so any race
                // in the other order risks dropping the first batch.
                prog.register(self.queue_id, &socket)?;
                let attachment = prog.attach(iface, flags).map_err(|e| {
                    if self.force_replace {
                        Error::Loader(format!(
                            "XDP attach to '{iface}' failed and atomic replacement of an \
                             existing program is not supported through the link API on this \
                             kernel (≥ 5.9). Remove the incumbent first \
                             (`ip link set dev {iface} xdp off`), then retry. Underlying: {e}"
                        ))
                    } else {
                        e
                    }
                })?;
                socket._xdp_attachment = Some(attachment);
            }
        }

        Ok(socket)
    }

    /// Build the XDP socket (stub without `af-xdp` feature).
    #[cfg(not(feature = "af-xdp"))]
    pub fn build(self) -> Result<XdpSocket, Error> {
        Err(Error::Config(
            "AF_XDP requires the 'af-xdp' feature flag".into(),
        ))
    }
}

// ── XdpSocket ────────────────────────────────────────────────────────────

/// AF_XDP socket handle.
///
/// Provides non-blocking `recv` / `send` / `flush` operations over a UMEM
/// region and 4 XDP rings (fill, RX, TX, completion).
///
/// Requires the `af-xdp` feature to construct via [`XdpSocketBuilder::build`].
///
/// `XdpSocket` is `Send` but **not** `Sync`. Pass it across threads if you
/// like, but only one thread at a time may call any method on it.
///
/// ```compile_fail
/// fn assert_sync<T: Sync>() {}
/// assert_sync::<netring::XdpSocket>();
/// ```
pub struct XdpSocket {
    #[cfg(feature = "af-xdp")]
    fd: OwnedFd,
    #[cfg(feature = "af-xdp")]
    umem: Umem,
    #[cfg(feature = "af-xdp")]
    fill: FillRing,
    #[cfg(feature = "af-xdp")]
    rx: RxRing,
    #[cfg(feature = "af-xdp")]
    tx: TxRing,
    #[cfg(feature = "af-xdp")]
    comp: CompletionRing,
    /// Set when the socket was bound with `XDP_USE_NEED_WAKEUP`. Determines
    /// whether `flush()` honors `tx.needs_wakeup()` or always kicks.
    #[cfg(feature = "af-xdp")]
    need_wakeup_enabled: bool,
    /// Issue #6 F2: whether the kernel bound this socket in zero-copy mode
    /// (`XDP_OPTIONS_ZEROCOPY`). Read once at build; surfaced via
    /// [`XdpSocket::is_zerocopy`].
    #[cfg(feature = "af-xdp")]
    zerocopy: bool,
    /// Issue #13: whether RX-metadata headroom was reserved, so the RX paths
    /// read the XDP metadata struct ahead of each frame. Off → zero overhead.
    #[cfg(feature = "af-xdp")]
    rx_metadata: bool,
    /// RAII guard holding the interface in promiscuous mode (set when built
    /// with [`XdpSocketBuilder::promiscuous`]). On drop, its fd closes and the
    /// kernel decrements `dev->promiscuity`, so the interface leaves
    /// promiscuous mode. `None` when promiscuous mode was not requested.
    #[cfg(feature = "af-xdp")]
    _promisc_guard: Option<crate::afpacket::socket::PromiscGuard>,
    /// RAII guard for an attached XDP program (if the socket was built
    /// with `with_default_program()`). Drops before the socket fd, so
    /// the program is detached from the interface before AF_XDP shuts
    /// down.
    #[cfg(feature = "xdp-loader")]
    _xdp_attachment: Option<loader::XdpAttachment>,
    // Without the feature, keep a private field so the struct is unconstructable.
    // `PhantomData<*const ()>` mirrors the with-feature `!Sync` property so
    // the `compile_fail` doctest that asserts `!Sync` works regardless of
    // which features the test run uses.
    #[cfg(not(feature = "af-xdp"))]
    _private: std::marker::PhantomData<*const ()>,
}

impl XdpSocket {
    /// Open an XDP socket on `interface` with default settings.
    ///
    /// Equivalent to `XdpSocketBuilder::default().interface(interface).build()`.
    /// Default mode is [`XdpMode::RxTx`] — for TX-only workloads use
    /// [`XdpSocketBuilder`] with [`XdpMode::Tx`].
    #[cfg(feature = "af-xdp")]
    pub fn open(interface: &str) -> Result<Self, Error> {
        XdpSocketBuilder::default().interface(interface).build()
    }

    /// Start building a configured XDP socket.
    #[cfg(feature = "af-xdp")]
    pub fn builder() -> XdpSocketBuilder {
        XdpSocketBuilder::default()
    }

    /// Take the next batch of packets as a zero-copy view (non-blocking).
    ///
    /// Returns `Some(batch)` borrowing from the UMEM region, or `None` if
    /// no packets are available right now. Pairs with
    /// [`Capture::next_batch`](crate::Capture) on the AF_PACKET side —
    /// same name, same semantics, same `Send`/`Sync` rules.
    ///
    /// The batch holds `&mut self`; only one batch can be live at a time.
    /// Dropping it returns its frames to the free list, releases the RX
    /// descriptors, and refills the fill ring.
    ///
    /// # Soundness — only one batch live at a time
    ///
    /// The batch's `&mut self` borrow is enforced by the compiler:
    ///
    /// ```compile_fail
    /// # fn _ex(mut s: netring::XdpSocket) {
    /// let b1 = s.next_batch();
    /// let b2 = s.next_batch();  // ERROR: two mutable borrows
    /// drop(b1);
    /// drop(b2);
    /// # }
    /// ```
    #[cfg(feature = "af-xdp")]
    pub fn next_batch(&mut self) -> Option<XdpBatch<'_>> {
        self.recycle_completed();
        self.rx
            .consumer_peek(64)
            .map(|tok| XdpBatch::new(self, tok))
    }

    /// Block until a batch is available, or `timeout` elapses.
    ///
    /// Mirrors [`PacketSource::next_batch_blocking`](crate::PacketSource)
    /// for the AF_XDP backend. Internally polls `POLLIN` (EINTR-safe) and
    /// retries [`next_batch`](Self::next_batch).
    #[cfg(feature = "af-xdp")]
    pub fn next_batch_blocking(
        &mut self,
        timeout: Duration,
    ) -> Result<Option<XdpBatch<'_>>, Error> {
        // Fast path: try non-blocking first. If a batch is already there
        // we skip the syscall entirely.
        //
        // We can't actually return early with `Some(batch)` without
        // re-entering the borrow checker — `self.next_batch()` borrows
        // `&mut self` for the lifetime of the returned XdpBatch, and we
        // can't both return that and fall through to poll if it was None.
        // Standard NLL workaround: separate paths.
        if !self.rx_is_empty() {
            return Ok(self.next_batch());
        }

        // No batch ready — poll.
        let mut pfds = [nix::poll::PollFd::new(
            self.fd.as_fd(),
            nix::poll::PollFlags::POLLIN,
        )];
        crate::syscall::poll_eintr_safe(&mut pfds, timeout).map_err(Error::Io)?;

        Ok(self.next_batch())
    }

    /// Internal: probe whether the RX ring has anything we could peek.
    /// Uses the cached producer index (no kernel sync) — false positives
    /// are fine (we'll just loop), false negatives just mean we go through
    /// poll() once unnecessarily on a fresh batch.
    #[cfg(feature = "af-xdp")]
    pub(crate) fn rx_is_empty(&self) -> bool {
        self.rx.cached_count() == 0
    }

    /// Whether the kernel bound this socket in **zero-copy** mode (issue #6 F2).
    ///
    /// `false` means COPY mode — the kernel copies each frame into the UMEM
    /// (works on any driver, including SKB/generic XDP, but caps throughput).
    /// For line rate on a real NIC, use a native-XDP driver with
    /// [`XdpFlags::DRV_MODE`](crate::xdp::XdpFlags).
    #[cfg(feature = "af-xdp")]
    pub fn is_zerocopy(&self) -> bool {
        self.zerocopy
    }

    /// Whether this socket reads NIC RX metadata, i.e. was built with
    /// [`XdpSocketBuilder::rx_metadata`] (issue #13). When `false`, every
    /// frame's [`rx_metadata`](crate::XdpPacket::rx_metadata) is absent.
    #[cfg(feature = "af-xdp")]
    pub fn rx_metadata_enabled(&self) -> bool {
        self.rx_metadata
    }

    /// Read + parse the RX-metadata struct from the headroom preceding the
    /// frame at `addr`. Returns absent metadata when disabled, when no program
    /// wrote it, or when the magic check fails (the degrade path).
    #[cfg(feature = "af-xdp")]
    fn read_rx_metadata(&self, addr: u64) -> flowscope::RxMetadata {
        if !self.rx_metadata {
            return flowscope::RxMetadata::default();
        }
        self.umem
            .data_before(addr, metadata::XdpRxMeta::LEN)
            .and_then(metadata::XdpRxMeta::from_headroom)
            .map(|m| m.rx_metadata())
            .unwrap_or_default()
    }

    /// Borrow the socket fd for polling (used by the multi-queue
    /// [`XdpCapture`](crate::xdp::XdpCapture) round-robin).
    #[cfg(all(feature = "af-xdp", feature = "xdp-loader"))]
    pub(crate) fn poll_fd(&self) -> std::os::fd::BorrowedFd<'_> {
        self.fd.as_fd()
    }

    /// Fresh (kernel-synced) RX readiness probe for the multi-queue round-robin.
    ///
    /// Unlike [`rx_is_empty`](Self::rx_is_empty) (which reads the *cached*
    /// producer index and so can't be used to *decide* whether to peek), this
    /// re-loads the kernel producer index, so `XdpCapture` can pick which
    /// socket to drain without consuming from the others.
    #[cfg(all(feature = "af-xdp", feature = "xdp-loader"))]
    pub(crate) fn rx_poll_ready(&mut self) -> bool {
        self.rx.refresh_count() > 0
    }

    /// Receive packets (non-blocking) as owned copies.
    ///
    /// Returns owned copies of received packets — convenient but allocates
    /// a `Vec<u8>` per packet plus the outer `Vec`. For zero-copy access
    /// use [`next_batch()`](Self::next_batch) instead.
    ///
    /// Returns an empty `Vec` if no packets are available.
    #[cfg(feature = "af-xdp")]
    pub fn recv(&mut self) -> Result<Vec<OwnedPacket>, Error> {
        // 1. Recycle completed TX frames
        self.recycle_completed();

        // 2. Peek RX ring
        let tok = match self.rx.consumer_peek(64) {
            Some(t) => t,
            None => return Ok(Vec::new()),
        };

        let mut packets = Vec::with_capacity(tok.n as usize);

        for i in 0..tok.n {
            let desc: libc::xdp_desc = self.rx.read_at(tok, i);
            let meta = self.read_rx_metadata(desc.addr);
            match self.umem.data_checked(desc.addr, desc.len as usize) {
                Some(data) => packets.push(OwnedPacket {
                    data: data.to_vec(),
                    timestamp: meta.hw_timestamp.unwrap_or_default(),
                    timestamp_clock: match meta.hw_timestamp {
                        Some(_) => crate::packet::TimestampClock::RawHardware,
                        None => crate::packet::TimestampClock::None,
                    },
                    original_len: desc.len as usize,
                    status: crate::packet::PacketStatus::default(),
                    direction: crate::packet::PacketDirection::Unknown(0),
                    rxhash: meta.rx_hash.map_or(0, |h| h.value),
                    vlan_tci: 0,
                    vlan_tpid: 0,
                    ll_protocol: 0,
                    source_ll_addr: [0u8; 8],
                    source_ll_addr_len: 0,
                }),
                None => {
                    // Defense in depth: a kernel that's misbehaving (or a
                    // shared-UMEM peer that wrote a corrupt desc) shouldn't
                    // panic the consumer.
                    tracing::warn!(
                        addr = desc.addr,
                        len = desc.len,
                        "AF_XDP: malformed RX descriptor; skipping"
                    );
                }
            }
            // Return frame to free list (will be refilled below) regardless.
            self.umem.free_frame(desc.addr);
        }

        // 3. Release consumed RX descriptors
        self.rx.consumer_release(tok);

        // 4. Refill fill ring with recycled frames
        self.refill();

        Ok(packets)
    }

    /// Send a raw packet (non-blocking).
    ///
    /// Copies `data` into a UMEM frame and submits a TX descriptor.
    /// Returns `Ok(true)` on success, `Ok(false)` if the TX ring or UMEM is full.
    ///
    /// Call [`flush`](Self::flush) after one or more `send` calls to kick the kernel.
    #[cfg(feature = "af-xdp")]
    pub fn send(&mut self, data: &[u8]) -> Result<bool, Error> {
        if data.len() > self.umem.frame_size() {
            return Err(Error::Config(format!(
                "packet {} bytes exceeds frame size {}",
                data.len(),
                self.umem.frame_size()
            )));
        }

        self.recycle_completed();

        let addr = match self.umem.alloc_frame() {
            Some(a) => a,
            None => return Ok(false),
        };

        // Copy data into UMEM frame.
        // The frame_size check at the top of send() ensures data.len() fits;
        // expect() here is for an internal invariant that should never fail.
        let buf = self
            .umem
            .data_mut_checked(addr, data.len())
            .expect("send: frame_size pre-check guarantees fit");
        buf.copy_from_slice(data);

        // Submit TX descriptor
        let tok = match self.tx.producer_reserve(1) {
            Some(t) => t,
            None => {
                self.umem.free_frame(addr);
                return Ok(false);
            }
        };
        self.tx.write_at(
            tok,
            0,
            libc::xdp_desc {
                addr,
                len: data.len() as u32,
                options: 0,
            },
        );
        self.tx.producer_submit(tok);

        Ok(true)
    }

    /// Flush pending TX frames by waking the kernel.
    ///
    /// When the socket was bound with `XDP_USE_NEED_WAKEUP` (the default), the
    /// kernel sets a flag in the TX ring whenever it actually needs a kick;
    /// we honor that and skip the syscall otherwise. For sockets bound without
    /// the flag, `needs_wakeup()` always returns false — we still kick.
    ///
    /// Uses `sendto(fd, NULL, 0, MSG_DONTWAIT, NULL, 0)`.
    /// `MSG_DONTWAIT` is **mandatory** — kernel returns `EOPNOTSUPP` without it.
    /// EINTR is retried; transient `EAGAIN`/`ENOBUFS` are reported as success.
    #[cfg(feature = "af-xdp")]
    pub fn flush(&self) -> Result<(), Error> {
        if self.need_wakeup_enabled && !self.tx.needs_wakeup() {
            return Ok(());
        }
        crate::syscall::sendto_kick_eintr_safe(self.fd.as_raw_fd(), libc::MSG_DONTWAIT)
            .map_err(Error::Io)
    }

    /// Poll for readability (incoming packets) with a timeout.
    ///
    /// Returns `true` if packets may be available. EINTR is handled internally.
    #[cfg(feature = "af-xdp")]
    pub fn poll(&self, timeout: Duration) -> Result<bool, Error> {
        // SAFETY: BorrowedFd is valid for the call; we only use it within the
        // duration of poll_eintr_safe.
        let fd = self.fd.as_fd();
        let mut pfds = [nix::poll::PollFd::new(fd, nix::poll::PollFlags::POLLIN)];
        let n = crate::syscall::poll_eintr_safe(&mut pfds, timeout).map_err(Error::Io)?;
        Ok(n > 0)
    }

    /// Get XDP socket statistics from the kernel.
    ///
    /// Counters are monotonically non-decreasing for the socket's lifetime
    /// (no destructive-read semantics).
    #[cfg(feature = "af-xdp")]
    pub fn statistics(&self) -> Result<XdpStats, Error> {
        socket::get_statistics(self.fd.as_fd()).map(XdpStats::from)
    }

    // ── Internal helpers ─────────────────────────────────────────────────

    /// Recycle frames from the completion ring back to the UMEM free list.
    #[cfg(feature = "af-xdp")]
    fn recycle_completed(&mut self) {
        let tok = match self.comp.consumer_peek(64) {
            Some(t) => t,
            None => return,
        };
        let mut addrs = [0u64; 64];
        for i in 0..tok.n {
            addrs[i as usize] = self.comp.read_at(tok, i);
        }
        self.umem.free_frames(&addrs[..tok.n as usize]);
        self.comp.consumer_release(tok);
    }

    /// Refill the fill ring with available UMEM frames.
    #[cfg(feature = "af-xdp")]
    fn refill(&mut self) {
        let want = self.umem.available().min(64) as u32;
        if want == 0 {
            return;
        }
        if let Some(tok) = self.fill.producer_reserve(want) {
            let mut filled = 0u32;
            for i in 0..tok.n {
                if let Some(addr) = self.umem.alloc_frame() {
                    self.fill.write_at(tok, i, addr);
                    filled += 1;
                }
            }
            // `want` was bounded by umem.available() so all reservations get
            // filled; the assertion documents the invariant.
            debug_assert_eq!(filled, tok.n);
            if filled > 0 {
                self.fill.producer_submit(tok);
            }
        }
    }
}

#[cfg(feature = "af-xdp")]
impl AsFd for XdpSocket {
    fn as_fd(&self) -> BorrowedFd<'_> {
        self.fd.as_fd()
    }
}

#[cfg(feature = "af-xdp")]
impl AsRawFd for XdpSocket {
    fn as_raw_fd(&self) -> std::os::fd::RawFd {
        self.fd.as_raw_fd()
    }
}

// SAFETY: All fields (OwnedFd, Umem, XdpRing) are Send. Access is mediated
// by &mut self on all operations.
#[cfg(feature = "af-xdp")]
unsafe impl Send for XdpSocket {}

impl std::fmt::Debug for XdpSocket {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        let mut d = f.debug_struct("XdpSocket");
        #[cfg(feature = "af-xdp")]
        {
            d.field("frame_size", &self.umem.frame_size());
            d.field("umem_available", &self.umem.available());
        }
        d.finish()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn builder_rejects_missing_interface() {
        let err = XdpSocketBuilder::default().build().unwrap_err();
        assert!(matches!(err, Error::Config(_)));
    }

    #[test]
    fn builder_defaults() {
        let b = XdpSocketBuilder::default();
        // Verify defaults via validate (fields are private)
        assert!(b.validate().is_err()); // no interface
        let b = b.interface("lo");
        assert!(b.validate().is_ok());
    }

    #[test]
    fn builder_validate_ok() {
        let b = XdpSocketBuilder::default().interface("lo");
        assert!(b.validate().is_ok());
    }

    #[test]
    fn builder_validate_zero_frame_size() {
        let b = XdpSocketBuilder::default().interface("lo").frame_size(0);
        assert!(b.validate().is_err());
    }

    #[test]
    fn builder_validate_zero_frame_count() {
        let b = XdpSocketBuilder::default().interface("lo").frame_count(0);
        assert!(b.validate().is_err());
    }

    #[test]
    fn builder_chaining() {
        let b = XdpSocketBuilder::default()
            .interface("eth0")
            .queue_id(3)
            .frame_size(2048)
            .frame_count(1024)
            .need_wakeup(false);
        assert_eq!(b.validate().unwrap(), "eth0");
    }

    #[test]
    fn builder_default_mode_is_rxtx() {
        let b = XdpSocketBuilder::default();
        assert_eq!(b.mode, XdpMode::RxTx);
    }

    #[test]
    fn builder_busy_poll_trio_chain() {
        let b = XdpSocketBuilder::default()
            .interface("lo")
            .queue_id(0)
            .busy_poll_us(50)
            .prefer_busy_poll(true)
            .busy_poll_budget(64);
        assert_eq!(b.busy_poll_us, Some(50));
        assert_eq!(b.prefer_busy_poll, Some(true));
        assert_eq!(b.busy_poll_budget, Some(64));
    }

    #[test]
    fn builder_busy_poll_default_unset() {
        let b = XdpSocketBuilder::default();
        assert_eq!(b.busy_poll_us, None);
        assert_eq!(b.prefer_busy_poll, None);
        assert_eq!(b.busy_poll_budget, None);
    }

    #[test]
    fn builder_mode_setter() {
        let b = XdpSocketBuilder::default().mode(XdpMode::Tx);
        assert_eq!(b.mode, XdpMode::Tx);

        let b = XdpSocketBuilder::default().mode(XdpMode::Custom { prefill: 256 });
        assert_eq!(b.mode, XdpMode::Custom { prefill: 256 });
    }

    #[test]
    fn builder_promiscuous_defaults_off_and_setter_toggles() {
        // Default is off — promiscuous mode is opt-in (issue #4).
        let b = XdpSocketBuilder::default();
        assert!(!b.promiscuous);

        let b = XdpSocketBuilder::default().promiscuous(true);
        assert!(b.promiscuous);

        // Idempotent / re-settable.
        let b = b.promiscuous(false);
        assert!(!b.promiscuous);
    }

    #[cfg(feature = "xdp-loader")]
    #[test]
    fn check_program_conflict_helper() {
        use loader::check_program_conflict;
        // Default state — neither set — is fine.
        assert!(check_program_conflict(false, false).is_ok());
        // Either set alone is fine.
        assert!(check_program_conflict(true, false).is_ok());
        assert!(check_program_conflict(false, true).is_ok());
        // Both set is the only conflict.
        let err = check_program_conflict(true, true).unwrap_err();
        assert!(format!("{err}").contains("conflicts"));
    }

    #[cfg(feature = "xdp-loader")]
    #[test]
    fn builder_with_default_program_chains() {
        let b = XdpSocketBuilder::default()
            .interface("lo")
            .with_default_program();
        assert!(b.attach_default);
        assert!(b.program.is_none());
    }

    /// Compute the prefill count the same way `build()` does, in isolation,
    /// so we can unit-test the policy without a live AF_XDP socket.
    fn compute_prefill(mode: XdpMode, available: usize, ring_size: usize) -> u32 {
        let cap_avail = available.min(ring_size);
        let n = match mode {
            XdpMode::Rx => cap_avail,
            XdpMode::Tx => 0,
            XdpMode::RxTx => cap_avail / 2,
            XdpMode::Custom { prefill } => (prefill as usize).min(cap_avail),
        };
        n as u32
    }

    #[test]
    fn prefill_tx_keeps_all_frames_in_free_list() {
        // The bug being fixed: prefill must not consume all frames in TX mode.
        assert_eq!(compute_prefill(XdpMode::Tx, 4096, 4096), 0);
    }

    #[test]
    fn prefill_rx_consumes_all_frames() {
        assert_eq!(compute_prefill(XdpMode::Rx, 4096, 4096), 4096);
    }

    #[test]
    fn prefill_rxtx_splits_in_half() {
        assert_eq!(compute_prefill(XdpMode::RxTx, 4096, 4096), 2048);
    }

    #[test]
    fn prefill_custom_clamped() {
        assert_eq!(
            compute_prefill(XdpMode::Custom { prefill: 100 }, 4096, 4096),
            100
        );
        // Clamps to ring_size when caller asks for too much.
        assert_eq!(
            compute_prefill(XdpMode::Custom { prefill: 8192 }, 4096, 4096),
            4096
        );
        // Clamps to available when caller asks for too much.
        assert_eq!(
            compute_prefill(XdpMode::Custom { prefill: 1000 }, 64, 4096),
            64
        );
    }
}