ferranet 0.2.0

//! The public, synchronous datalink channel: [`Channel`], [`ChannelBuilder`], [`Sender`], and
//! [`Receiver`].
//!
//! A channel is opened with the builder and split into a [`Sender`] and a [`Receiver`], each
//! owning its own socket and file descriptor. By default the channel uses zero-copy
//! `PACKET_MMAP` rings; [`ChannelBuilder::basic`] selects a simpler `send`/`recv` backend.

use std::os::fd::{AsFd, BorrowedFd, OwnedFd};

use crate::block::{Block, Frame, FrameMeta, PacketType};
use crate::dummy::{RxQueue, SentQueue, drain_readiness};
use crate::error::{Error, Result};
use crate::interface::{MacAddr, index_of};
use crate::sys::Stats;
use crate::sys::linux::ring::{RxRing, TxRing};
use crate::sys::linux::socket::{ETH_P_ALL, PacketSocket, SocketKind, poll_readable, poll_writable};
use crate::sys::linux::tpacket::{RingLayout, min_frame_size, tpacket_align};

/// The backend a channel uses to move frames.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum Mode {
    /// Zero-copy `PACKET_MMAP` rings (TPACKET_V3 RX / V2 TX).
    Ring,
    /// Plain `send`/`recv` syscalls with a single-copy receive.
    Basic,
}

/// How the kernel distributes received frames across the sockets in a `PACKET_FANOUT` group.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
#[non_exhaustive]
pub enum FanoutMode {
    /// Hash over the flow (src/dst/proto) so a flow always lands on the same socket. The default.
    Hash,
    /// Round-robin across the sockets, ignoring flow affinity.
    LoadBalance,
    /// Steer by the CPU that processed the packet.
    Cpu,
    /// Send to the first socket, rolling over to the next only when it is backlogged.
    RollOver,
    /// Pseudo-random socket selection.
    Random,
    /// Steer by the NIC's hardware receive-queue mapping.
    QueueMap,
}

impl FanoutMode {
    fn raw(self) -> u16 {
        let v = match self {
            FanoutMode::Hash => libc::PACKET_FANOUT_HASH,
            FanoutMode::LoadBalance => libc::PACKET_FANOUT_LB,
            FanoutMode::Cpu => libc::PACKET_FANOUT_CPU,
            FanoutMode::RollOver => libc::PACKET_FANOUT_ROLLOVER,
            FanoutMode::Random => libc::PACKET_FANOUT_RND,
            FanoutMode::QueueMap => libc::PACKET_FANOUT_QM,
        };
        v as u16
    }
}

/// Geometry for a `PACKET_MMAP` ring.
///
/// The defaults (an 8 MiB RX ring / 4 MiB TX ring of 2 KiB frames) suit high-rate use; for
/// constrained hardware see [`RingConfig::small`] and [`RingConfig::snaplen`].
#[derive(Debug, Clone, Copy)]
pub struct RingConfig {
    /// Size of each block in bytes (a multiple of the page size and of `frame_size`).
    pub block_size: u32,
    /// Number of blocks in the ring.
    pub block_count: u32,
    /// Size of each frame slot in bytes (a multiple of 16, at least an MTU plus header).
    pub frame_size: u32,
    /// v3 block-retire timeout in milliseconds (ignored for the TX ring).
    pub retire_blk_tov_ms: u32,
    /// If set, capture only roughly this many bytes per frame (a snap length), overriding
    /// `frame_size` with a small power-of-two derived from it.
    pub snaplen: Option<u32>,
}

impl RingConfig {
    /// The default receive geometry: 8 × 1 MiB blocks, 2 KiB frames, 60 ms retire (≈ 8 MiB).
    #[must_use]
    pub const fn rx_default() -> Self {
        RingConfig {
            block_size: 1 << 20,
            block_count: 8,
            frame_size: 2048,
            retire_blk_tov_ms: 60,
            snaplen: None,
        }
    }

    /// The default transmit geometry: 4 × 1 MiB blocks, 2 KiB frames (≈ 4 MiB).
    #[must_use]
    pub const fn tx_default() -> Self {
        RingConfig {
            block_size: 1 << 20,
            block_count: 4,
            frame_size: 2048,
            retire_blk_tov_ms: 0,
            snaplen: None,
        }
    }

    /// A small-footprint geometry for constrained hardware: 4 × 64 KiB blocks (≈ 256 KiB), which
    /// stays cache-resident and costs ~32× less RAM than the default while remaining ample at low
    /// packet rates. Combine with [`RingConfig::snaplen`] for header-only capture.
    #[must_use]
    pub const fn small() -> Self {
        RingConfig {
            block_size: 1 << 16,
            block_count: 4,
            frame_size: 2048,
            retire_blk_tov_ms: 60,
            snaplen: None,
        }
    }

    /// Captures only roughly the first `bytes` of each frame.
    ///
    /// The kernel truncates its copy into the ring, so far less memory is moved and many more
    /// frames pack into each block — a large win for header-only monitoring on slow hardware. The
    /// captured length is rounded up to a power-of-two frame size; [`Frame::wire_len`] still
    /// reports the true on-wire length.
    #[must_use]
    pub const fn snaplen(mut self, bytes: u32) -> Self {
        self.snaplen = Some(bytes);
        self
    }

    fn to_layout(self) -> Result<RingLayout> {
        let frame_size = match self.snaplen {
            Some(snap) => frame_size_for_snaplen(snap).ok_or_else(|| {
                Error::invalid_config(format!("snaplen {snap} is too large for a ring frame"))
            })?,
            None => self.frame_size,
        };
        RingLayout::new(
            self.block_size,
            self.block_count,
            frame_size,
            self.retire_blk_tov_ms,
            page_size(),
        )
    }
}

/// Derives a frame size that captures at least `snaplen` bytes of payload, or `None` if the snap
/// length is too large to fit any power-of-two frame.
///
/// A power of two is chosen so it cleanly divides the (power-of-two) block sizes and is a multiple
/// of `TPACKET_ALIGNMENT`. With `snaplen = 1500` this lands on the 2 KiB default; small snap
/// lengths give much smaller, denser frames (e.g. 128 → 256-byte frames).
fn frame_size_for_snaplen(snaplen: u32) -> Option<u32> {
    // Header + room for the link-layer offset/alignment the kernel places before the payload.
    let overhead = tpacket_align(libc::TPACKET3_HDRLEN) as u32 + 32;
    let needed = overhead.checked_add(snaplen)?.max(min_frame_size() as u32);
    needed.checked_next_power_of_two().map(|v| v.max(128))
}

fn page_size() -> u32 {
    // SAFETY: sysconf with a valid name has no preconditions and no side effects.
    let v = unsafe { libc::sysconf(libc::_SC_PAGESIZE) };
    if v <= 0 { 4096 } else { v as u32 }
}

/// Builder for a datalink [`Channel`].
///
/// Create one with [`Channel::builder`], configure it, then call [`ChannelBuilder::build_sync`]
/// (or `build_async` with the `tokio` feature) to obtain a [`Sender`]/[`Receiver`] pair.
#[derive(Debug, Clone)]
pub struct ChannelBuilder {
    interface: String,
    protocol: u16,
    kind: SocketKind,
    mode: Mode,
    promiscuous: bool,
    rx: RingConfig,
    tx: RingConfig,
    fanout: FanoutMode,
    fanout_group: Option<u16>,
}

impl ChannelBuilder {
    fn new(interface: impl Into<String>) -> Self {
        ChannelBuilder {
            interface: interface.into(),
            protocol: ETH_P_ALL,
            kind: SocketKind::Raw,
            mode: Mode::Ring,
            promiscuous: false,
            rx: RingConfig::rx_default(),
            tx: RingConfig::tx_default(),
            fanout: FanoutMode::Hash,
            fanout_group: None,
        }
    }

    /// Sets the EtherType to bind to, in host byte order (default: `ETH_P_ALL`, every protocol).
    #[must_use]
    pub fn protocol(mut self, protocol: u16) -> Self {
        self.protocol = protocol;
        self
    }

    /// Selects `SOCK_DGRAM`: the kernel strips/adds the link-layer header for you (default is
    /// `SOCK_RAW`, where frames include the full Ethernet header).
    #[must_use]
    pub fn dgram(mut self) -> Self {
        self.kind = SocketKind::Dgram;
        self
    }

    /// Uses the simple `send`/`recv` backend instead of zero-copy `PACKET_MMAP` rings.
    #[must_use]
    pub fn basic(mut self) -> Self {
        self.mode = Mode::Basic;
        self
    }

    /// Enables promiscuous mode on the interface.
    #[must_use]
    pub fn promiscuous(mut self, on: bool) -> Self {
        self.promiscuous = on;
        self
    }

    /// Sets the load-balancing strategy used by [`ChannelBuilder::build_fanout_rx`].
    #[must_use]
    pub fn fanout(mut self, mode: FanoutMode) -> Self {
        self.fanout = mode;
        self
    }

    /// Pins the fanout group to a specific id (default: a fresh kernel-allocated unique id).
    ///
    /// Use this to deliberately join a group shared across processes. Note the kernel merges any
    /// same-id, same-mode joins in a namespace into one load-balanced group, so a pinned id must
    /// be coordinated between its users.
    #[must_use]
    pub fn fanout_group(mut self, id: u16) -> Self {
        self.fanout_group = Some(id);
        self
    }

    /// Overrides the receive ring geometry.
    #[must_use]
    pub fn rx_ring(mut self, config: RingConfig) -> Self {
        self.rx = config;
        self
    }

    /// Overrides the transmit ring geometry.
    #[must_use]
    pub fn tx_ring(mut self, config: RingConfig) -> Self {
        self.tx = config;
        self
    }

    /// Opens the channel synchronously, returning a transmit/receive pair.
    pub fn build_sync(&self) -> Result<(Sender, Receiver)> {
        let parts = self.open()?;
        Ok((Sender { inner: parts.tx }, Receiver::new(parts.rx)))
    }

    /// Opens the two bound sockets and constructs the backend halves.
    pub(crate) fn open(&self) -> Result<ChannelParts> {
        let idx = index_of(&self.interface)?;
        Ok(ChannelParts { rx: self.open_rx_backend(idx)?, tx: self.open_tx_backend(idx)? })
    }

    fn open_rx_backend(&self, idx: crate::IfIndex) -> Result<RxBackend> {
        let sock = PacketSocket::open(idx, self.protocol, self.kind)?;
        if self.promiscuous {
            sock.set_promiscuous(true)?;
        }
        match self.mode {
            Mode::Ring => Ok(RxBackend::Ring(RxRing::open(sock, self.rx.to_layout()?)?)),
            Mode::Basic => Ok(RxBackend::Basic { sock, buf: vec![0u8; 65_536] }),
        }
    }

    fn open_tx_backend(&self, idx: crate::IfIndex) -> Result<TxBackend> {
        let sock = PacketSocket::open(idx, self.protocol, self.kind)?;
        match self.mode {
            Mode::Ring => Ok(TxBackend::Ring(TxRing::open(sock, self.tx.to_layout()?)?)),
            Mode::Basic => Ok(TxBackend::Basic(sock)),
        }
    }

    /// Builds `count` receive backends joined to a single `PACKET_FANOUT` group. Shared by the
    /// sync and async fanout entry points.
    pub(crate) fn build_fanout_backends(&self, count: usize) -> Result<Vec<RxBackend>> {
        if count == 0 {
            return Err(Error::invalid_config("fanout receiver count must be at least 1"));
        }
        let idx = index_of(&self.interface)?;
        let mode = self.fanout.raw();
        let mut backends = Vec::with_capacity(count);

        // The first socket determines the group id: a pinned one if configured, otherwise a
        // kernel-allocated unique id, so two unrelated processes can never guess their way into
        // one shared (frame-stealing) group. The remaining sockets then join that id.
        let first = self.open_rx_backend(idx)?;
        let group = match self.fanout_group {
            Some(id) => {
                first.set_fanout(id, mode)?;
                id
            }
            None => first.set_fanout_unique(mode, default_group_id())?,
        };
        backends.push(first);
        for _ in 1..count {
            let backend = self.open_rx_backend(idx)?;
            backend.set_fanout(group, mode)?;
            backends.push(backend);
        }
        Ok(backends)
    }

    /// Opens `count` receivers load-balanced across a `PACKET_FANOUT` group for multi-core RX.
    ///
    /// Move each [`Receiver`] onto its own thread (ideally pinned to a core) and call
    /// [`Receiver::recv_block`] on it; the kernel distributes incoming frames across the group
    /// according to the configured [`FanoutMode`]. Fanout is receive-only; build a normal channel
    /// for transmit.
    pub fn build_fanout_rx(&self, count: usize) -> Result<Vec<Receiver>> {
        Ok(self
            .build_fanout_backends(count)?
            .into_iter()
            .map(Receiver::new)
            .collect())
    }
}

/// Fallback fanout group id for kernels without `PACKET_FANOUT_FLAG_UNIQUEID` (< 4.5): the low
/// bits of the PID plus a per-call counter, so multiple groups in one process don't collide.
/// Unrelated processes can still collide in this scheme; on modern kernels the kernel-allocated
/// unique id is used instead. Override with [`ChannelBuilder::fanout_group`].
fn default_group_id() -> u16 {
    use std::sync::atomic::{AtomicU16, Ordering};
    static SEQ: AtomicU16 = AtomicU16::new(0);
    // SAFETY: getpid has no preconditions and no side effects.
    let pid = unsafe { libc::getpid() } as u16;
    pid.wrapping_add(SEQ.fetch_add(1, Ordering::Relaxed))
}

/// Internal: the two halves produced when opening a channel.
pub(crate) struct ChannelParts {
    pub(crate) rx: RxBackend,
    pub(crate) tx: TxBackend,
}

#[derive(Debug)]
pub(crate) enum RxBackend {
    Ring(RxRing),
    Basic { sock: PacketSocket, buf: Vec<u8> },
    /// In-memory test backend (see [`crate::dummy`]).
    Dummy { queue: RxQueue, readiness: OwnedFd, last: Vec<u8> },
}

impl RxBackend {
    #[cfg(feature = "tokio")]
    pub(crate) fn borrow_fd(&self) -> BorrowedFd<'_> {
        match self {
            RxBackend::Ring(ring) => ring.as_fd(),
            RxBackend::Basic { sock, .. } => sock.as_fd(),
            RxBackend::Dummy { readiness, .. } => readiness.as_fd(),
        }
    }

    pub(crate) fn set_fanout(&self, group_id: u16, mode: u16) -> Result<()> {
        match self {
            RxBackend::Ring(ring) => ring.set_fanout(group_id, mode),
            RxBackend::Basic { sock, .. } => sock.set_fanout(group_id, mode),
            RxBackend::Dummy { .. } => Ok(()),
        }
    }

    /// Claims a kernel-allocated unique fanout group, returning its id (see
    /// [`PacketSocket::set_fanout_unique`]).
    pub(crate) fn set_fanout_unique(&self, mode: u16, fallback_id: u16) -> Result<u16> {
        match self {
            RxBackend::Ring(ring) => ring.set_fanout_unique(mode, fallback_id),
            RxBackend::Basic { sock, .. } => sock.set_fanout_unique(mode, fallback_id),
            RxBackend::Dummy { .. } => Ok(fallback_id),
        }
    }
}

#[derive(Debug)]
pub(crate) enum TxBackend {
    Ring(TxRing),
    Basic(PacketSocket),
    /// In-memory test backend (see [`crate::dummy`]).
    Dummy { sent: SentQueue, fd: OwnedFd },
}

impl TxBackend {
    #[cfg(feature = "tokio")]
    pub(crate) fn borrow_fd(&self) -> BorrowedFd<'_> {
        match self {
            TxBackend::Ring(tx) => tx.as_fd(),
            TxBackend::Basic(sock) => sock.as_fd(),
            TxBackend::Dummy { fd, .. } => fd.as_fd(),
        }
    }

    /// Attempts to send one frame without blocking; returns `WouldBlock` if the queue is full.
    pub(crate) fn try_send(&mut self, frame: &[u8]) -> std::io::Result<usize> {
        use crate::sys::RawChannel;
        match self {
            TxBackend::Ring(tx) => tx.send(frame),
            TxBackend::Basic(sock) => sock.send(frame),
            TxBackend::Dummy { sent, .. } => {
                sent.lock().expect("dummy queue poisoned").push_back(frame.to_vec());
                Ok(frame.len())
            }
        }
    }

    /// Attempts to send a batch without blocking; returns the number of frames accepted.
    pub(crate) fn try_send_batch(&mut self, frames: &[&[u8]]) -> std::io::Result<usize> {
        use crate::sys::RawChannel;
        match self {
            TxBackend::Ring(tx) => tx.send_batch(frames),
            TxBackend::Basic(sock) => {
                let mut sent = 0;
                for frame in frames {
                    match sock.send(frame) {
                        Ok(_) => sent += 1,
                        Err(e) if sent > 0 && e.kind() == std::io::ErrorKind::WouldBlock => break,
                        Err(e) => return Err(e),
                    }
                }
                Ok(sent)
            }
            TxBackend::Dummy { sent, .. } => {
                let mut q = sent.lock().expect("dummy queue poisoned");
                for frame in frames {
                    q.push_back(frame.to_vec());
                }
                Ok(frames.len())
            }
        }
    }

    /// The maximum frame length, if the backend imposes one (ring backend only).
    pub(crate) fn max_frame_len(&self) -> Option<usize> {
        match self {
            TxBackend::Ring(tx) => Some(tx.max_frame_len()),
            TxBackend::Basic(_) | TxBackend::Dummy { .. } => None,
        }
    }

    /// Whether accepted frames are still waiting for the kernel to pick them up. Only the ring
    /// backend defers transmission this way. (The sync path checks inside `finish_kick`; only
    /// the async sender needs the standalone query.)
    #[cfg(feature = "tokio")]
    pub(crate) fn has_unsent(&self) -> bool {
        match self {
            TxBackend::Ring(tx) => tx.has_unsent(),
            TxBackend::Basic(_) | TxBackend::Dummy { .. } => false,
        }
    }

    /// Blocking follow-up kick for frames the kernel has not picked up (see
    /// [`TxRing::finish_kick`]).
    pub(crate) fn finish_kick(&self) -> std::io::Result<()> {
        match self {
            TxBackend::Ring(tx) => tx.finish_kick(),
            TxBackend::Basic(_) | TxBackend::Dummy { .. } => Ok(()),
        }
    }
}

/// Entry point for opening a datalink channel.
///
/// This type is purely a namespace for [`Channel::builder`]; the working objects are the
/// [`Sender`] and [`Receiver`] it produces.
#[derive(Debug)]
pub struct Channel;

impl Channel {
    /// Starts building a channel on the named interface (e.g. `"eth0"`).
    #[must_use]
    pub fn builder(interface: impl Into<String>) -> ChannelBuilder {
        ChannelBuilder::new(interface)
    }
}

/// The transmit half of a datalink channel.
#[derive(Debug)]
pub struct Sender {
    inner: TxBackend,
}

impl Sender {
    /// Builds a sender over the in-memory test backend (see [`crate::dummy`]).
    pub(crate) fn from_dummy(sent: SentQueue, fd: OwnedFd) -> Self {
        Sender { inner: TxBackend::Dummy { sent, fd } }
    }

    /// Sends a single raw frame, blocking until the ring/queue has transmit capacity.
    ///
    /// On return the frame has been handed to the kernel's transmit path (waiting out a full send
    /// buffer if needed) — unless the kernel previously rejected another frame, in which case
    /// delivery of frames queued behind the rejected slot resumes as the ring wraps.
    ///
    /// For `SOCK_RAW` channels the frame must include the full Ethernet header.
    pub fn send(&mut self, frame: &[u8]) -> Result<()> {
        if let Some(max) = self.inner.max_frame_len() {
            if frame.len() > max {
                return Err(Error::FrameTooLarge { len: frame.len(), max });
            }
        }
        loop {
            match self.inner.try_send(frame) {
                Ok(_) => break,
                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                    // Bounded wait: POLLOUT tracks only the kernel head slot, so it can stay
                    // false while other slots free up; time out and re-try rather than trust it.
                    poll_writable(self.as_fd(), 100).map_err(Error::Send)?;
                    self.inner.finish_kick().map_err(Error::Send)?;
                }
                Err(e) => return Err(Error::Send(e)),
            }
        }
        self.inner.finish_kick().map_err(Error::Send)
    }

    /// Sends a batch of frames, blocking until at least one is accepted; returns how many were
    /// accepted. The same hand-off guarantee as [`Sender::send`] applies to the accepted frames.
    ///
    /// With a TX ring this fills many slots before a single flushing syscall; the basic backend
    /// sends them one at a time.
    pub fn send_batch(&mut self, frames: &[&[u8]]) -> Result<usize> {
        if frames.is_empty() {
            return Ok(0);
        }
        check_batch_frame_sizes(frames, self.inner.max_frame_len())?;
        let sent = loop {
            match self.inner.try_send_batch(frames) {
                Ok(n) => break n,
                Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                    poll_writable(self.as_fd(), 100).map_err(Error::Send)?;
                    self.inner.finish_kick().map_err(Error::Send)?;
                }
                Err(e) => return Err(Error::Send(e)),
            }
        };
        self.inner.finish_kick().map_err(Error::Send)?;
        Ok(sent)
    }

    /// Borrows the underlying file descriptor (e.g. for advanced configuration or interop).
    pub fn as_fd(&self) -> BorrowedFd<'_> {
        match &self.inner {
            TxBackend::Ring(tx) => tx.as_fd(),
            TxBackend::Basic(sock) => sock.as_fd(),
            TxBackend::Dummy { fd, .. } => fd.as_fd(),
        }
    }
}

/// The receive half of a datalink channel.
#[derive(Debug)]
pub struct Receiver {
    inner: RxBackend,
    /// Running cumulative counters; the kernel's PACKET_STATISTICS is reset-on-read, so we fold
    /// each delta in here to present monotonic totals.
    stats_total: Stats,
}

impl Receiver {
    pub(crate) fn new(inner: RxBackend) -> Self {
        Receiver { inner, stats_total: Stats::default() }
    }

    /// Builds a receiver over the in-memory test backend (see [`crate::dummy`]).
    pub(crate) fn from_dummy(queue: RxQueue, readiness: OwnedFd) -> Self {
        Receiver::new(RxBackend::Dummy { queue, readiness, last: Vec::new() })
    }

    /// Receives the next batch of frames, blocking until at least one is available.
    ///
    /// The returned [`Block`] borrows zero-copy from kernel memory (ring backend) or from an
    /// internal buffer (basic backend); it must be dropped before the next call.
    pub fn recv_block(&mut self) -> Result<Block<'_>> {
        match &mut self.inner {
            RxBackend::Ring(ring) => {
                loop {
                    if ring.block_ready() {
                        break;
                    }
                    poll_readable(ring.as_fd(), -1).map_err(Error::Recv)?;
                }
                Ok(ring.recv_block().expect("block reported ready"))
            }
            RxBackend::Basic { sock, buf } => {
                let (n, pkttype) = loop {
                    match sock.recv_into(buf) {
                        Ok(v) => break v,
                        Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => {
                            poll_readable(sock.as_fd(), -1).map_err(Error::Recv)?;
                        }
                        Err(e) => return Err(Error::Recv(e)),
                    }
                };
                let captured = n.min(buf.len());
                let meta = FrameMeta {
                    wire_len: n,
                    timestamp: None,
                    vlan: None,
                    packet_type: PacketType::from_raw(pkttype),
                };
                Ok(Block::single(Some(Frame::new(&buf[..captured], meta))))
            }
            RxBackend::Dummy { queue, readiness, last } => {
                // Block until a frame is queued (signalled via the pipe), then deliver it. If the
                // injector was dropped and the queue is empty, the network is closed.
                let frame = loop {
                    let popped = queue.lock().expect("dummy queue poisoned").pop_front();
                    match popped {
                        Some(Ok(v)) => break v,
                        Some(Err(e)) => return Err(Error::Recv(e)),
                        None => {
                            poll_readable(readiness.as_fd(), -1).map_err(Error::Recv)?;
                            let eof = drain_readiness(readiness.as_fd());
                            if eof && queue.lock().expect("dummy queue poisoned").is_empty() {
                                return Err(Error::Recv(std::io::Error::from(
                                    std::io::ErrorKind::UnexpectedEof,
                                )));
                            }
                        }
                    }
                };
                *last = frame;
                let meta = FrameMeta {
                    wire_len: last.len(),
                    timestamp: None,
                    vlan: None,
                    packet_type: dummy_packet_type(last),
                };
                Ok(Block::single(Some(Frame::new(&last[..], meta))))
            }
        }
    }

    /// Returns cumulative receive/drop statistics for this channel since it was opened.
    ///
    /// `dropped` counts packets the kernel discarded because the ring was full (data loss);
    /// `freezes` counts ring-overflow events. Reading stats also clears the kernel's
    /// `TP_STATUS_LOSING` flag, so pair this with [`Block::is_losing`](crate::Block::is_losing):
    /// the block flag tells you *that* you are losing, this tells you *how much*.
    pub fn stats(&mut self) -> Result<Stats> {
        let delta = match &mut self.inner {
            RxBackend::Ring(ring) => ring.stats().map_err(Error::Stats)?,
            RxBackend::Basic { sock, .. } => sock.statistics().map_err(Error::Stats)?,
            RxBackend::Dummy { .. } => Stats::default(),
        };
        self.stats_total.accumulate(delta);
        Ok(self.stats_total)
    }

    /// Enables or disables promiscuous mode on the interface.
    pub fn set_promiscuous(&self, on: bool) -> Result<()> {
        match &self.inner {
            RxBackend::Ring(ring) => ring.set_promiscuous(on),
            RxBackend::Basic { sock, .. } => sock.set_promiscuous(on),
            RxBackend::Dummy { .. } => Ok(()),
        }
    }

    /// Borrows the underlying file descriptor (e.g. for advanced configuration or interop).
    pub fn as_fd(&self) -> BorrowedFd<'_> {
        match &self.inner {
            RxBackend::Ring(ring) => ring.as_fd(),
            RxBackend::Basic { sock, .. } => sock.as_fd(),
            RxBackend::Dummy { readiness, .. } => readiness.as_fd(),
        }
    }
}

/// Rejects a batch containing a frame over the backend's limit up front, so the condition
/// surfaces as [`Error::FrameTooLarge`] — the same error `send` reports — rather than as a raw
/// I/O error from whichever slot-fill happens to hit it.
pub(crate) fn check_batch_frame_sizes(frames: &[&[u8]], max: Option<usize>) -> Result<()> {
    if let Some(max) = max {
        if let Some(f) = frames.iter().find(|f| f.len() > max) {
            return Err(Error::FrameTooLarge { len: f.len(), max });
        }
    }
    Ok(())
}

/// Best-effort packet-type classification for a dummy frame, from its destination MAC.
pub(crate) fn dummy_packet_type(frame: &[u8]) -> PacketType {
    frame
        .get(0..6)
        .map(|b| PacketType::from_dest_mac(MacAddr([b[0], b[1], b[2], b[3], b[4], b[5]])))
        .unwrap_or(PacketType::Other(0))
}

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

    #[test]
    fn builder_defaults() {
        let b = Channel::builder("eth0");
        assert_eq!(b.protocol, ETH_P_ALL);
        assert_eq!(b.mode, Mode::Ring);
        assert!(!b.promiscuous);
    }

    #[test]
    fn builder_is_chainable() {
        let b = Channel::builder("eth0").dgram().basic().promiscuous(true).protocol(0x0800);
        assert_eq!(b.mode, Mode::Basic);
        assert!(b.promiscuous);
        assert_eq!(b.protocol, 0x0800);
        assert_eq!(b.kind, SocketKind::Dgram);
    }

    #[test]
    #[cfg_attr(miri, ignore = "to_layout reads the page size via sysconf")]
    fn ring_config_validates() {
        assert!(RingConfig::rx_default().to_layout().is_ok());
        assert!(RingConfig::tx_default().to_layout().is_ok());
        let bad = RingConfig { block_size: 100, ..RingConfig::rx_default() };
        assert!(bad.to_layout().is_err());
    }

    #[test]
    fn fanout_zero_receivers_is_invalid_config() {
        // The count check precedes any syscall, so this must fail fast with InvalidConfig even
        // for a nonexistent interface (and runs privilege-free, including under Miri).
        let r = Channel::builder("definitely-not-an-iface-xyz").build_fanout_rx(0);
        assert!(matches!(r, Err(Error::InvalidConfig(_))));
    }

    #[test]
    #[cfg_attr(miri, ignore = "calls if_nametoindex")]
    fn build_on_missing_interface_errors() {
        let r = Channel::builder("definitely-not-an-iface-xyz").build_sync();
        assert!(matches!(r, Err(Error::InterfaceNotFound(_))));
    }

    #[test]
    #[cfg_attr(miri, ignore = "to_layout reads the page size via sysconf")]
    fn small_preset_is_compact_and_valid() {
        let layout = RingConfig::small().to_layout().expect("small layout valid");
        assert_eq!(layout.map_len(), (1usize << 16) * 4); // 256 KiB, vs 8 MiB default
    }

    #[test]
    #[cfg_attr(miri, ignore = "to_layout reads the page size via sysconf")]
    fn absurd_snaplen_is_invalid_config_not_panic() {
        // Near-u32::MAX snap lengths used to overflow next_power_of_two (a panic in debug
        // builds); they must surface as configuration errors naming the offending value.
        let r = RingConfig::rx_default().snaplen(u32::MAX).to_layout();
        assert!(matches!(r, Err(Error::InvalidConfig(_))));
        let msg = r.unwrap_err().to_string();
        assert!(msg.contains(&u32::MAX.to_string()), "message should name the value: {msg}");
        let r = RingConfig::rx_default().snaplen((1 << 31) - 1).to_layout();
        assert!(matches!(r, Err(Error::InvalidConfig(_))));
    }

    #[test]
    #[cfg_attr(miri, ignore = "to_layout reads the page size via sysconf")]
    fn snaplen_derives_dense_power_of_two_frames() {
        assert_eq!(frame_size_for_snaplen(1500), Some(2048)); // full frame ≈ the default
        assert_eq!(frame_size_for_snaplen(128), Some(256)); // header-only: 8× denser
        assert!(frame_size_for_snaplen(1).unwrap() >= 128);
        assert!(frame_size_for_snaplen(64).unwrap().is_power_of_two());

        let layout = RingConfig::small().snaplen(128).to_layout().expect("snaplen layout valid");
        assert_eq!(layout.frame_size, 256);
        // 256 KiB / 256-byte frames packs many more frames than the 2 KiB default.
        assert_eq!(layout.frame_count, layout.map_len() as u32 / 256);
    }

    proptest::proptest! {
        /// Any snap length yields a power-of-two frame size that is at least the minimum and
        /// cleanly divides the default 1 MiB block (so the derived geometry always validates).
        #[test]
        #[cfg_attr(miri, ignore = "proptest is slow under Miri and covers safe arithmetic")]
        fn snaplen_frame_size_invariants(snaplen in 0u32..=65535) {
            let fs = frame_size_for_snaplen(snaplen).expect("sane snaplen always fits");
            proptest::prop_assert!(fs.is_power_of_two());
            proptest::prop_assert!(fs >= 128);
            proptest::prop_assert!(fs as usize >= min_frame_size());
            proptest::prop_assert_eq!((1u32 << 20) % fs, 0);
        }
    }
}