ferranet 0.2.0

//! End-to-end datalink tests over a `veth` pair.
//!
//! These are `#[ignore]`d by default because opening `AF_PACKET` sockets needs `CAP_NET_RAW` and
//! creating interfaces needs `CAP_NET_ADMIN`. Run them inside an unprivileged user + network
//! namespace, which grants both without real root:
//!
//! ```sh
//! unshare --user --map-root-user --net \
//!     env RUSTC_WRAPPER= cargo test --test veth -- --ignored --test-threads=1
//! ```
//!
//! Each test builds its own uniquely-named `veth` pair, sends a tagged frame on one end, and
//! asserts it is received (zero-copy) on the other.

use std::sync::Arc;
use std::sync::atomic::{AtomicBool, Ordering};
use std::thread;
use std::time::{Duration, Instant};

use ferranet::{Channel, FanoutMode, RingConfig};

mod common;
use common::{make_frame, run, wait_readable};

/// A `veth` pair that deletes itself on drop.
struct VethPair {
    a: String,
    b: String,
}

impl VethPair {
    fn new(a: &str, b: &str) -> Self {
        // Clean up any stale interfaces from a previous aborted run, then create fresh ones.
        let _ = run("ip", &["link", "del", a]);
        assert!(
            run("ip", &["link", "add", a, "type", "veth", "peer", "name", b]),
            "failed to create veth pair (need CAP_NET_ADMIN — run inside a netns)"
        );
        assert!(run("ip", &["link", "set", a, "up"]), "failed to bring {a} up");
        assert!(run("ip", &["link", "set", b, "up"]), "failed to bring {b} up");
        VethPair { a: a.to_owned(), b: b.to_owned() }
    }
}

impl Drop for VethPair {
    fn drop(&mut self) {
        let _ = run("ip", &["link", "del", &self.a]);
    }
}

fn rx_config() -> RingConfig {
    // A short block-retire timeout keeps the test snappy: the kernel closes a partially-filled
    // block (making our frame visible) after this many milliseconds.
    RingConfig { retire_blk_tov_ms: 10, ..RingConfig::rx_default() }
}

#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn ring_sync_roundtrip() {
    let veth = VethPair::new("fnvr0", "fnvr1");
    let (mut tx, _) = Channel::builder(&veth.a).build_sync().expect("open sender");
    let (_, mut rx) =
        Channel::builder(&veth.b).rx_ring(rx_config()).build_sync().expect("open receiver");

    let payload = b"ferranet-ring-sync";
    let frame = make_frame(payload);
    tx.send(&frame).expect("send frame");

    let mut found = false;
    let deadline = std::time::Instant::now() + Duration::from_secs(3);
    'outer: while std::time::Instant::now() < deadline {
        if !wait_readable(rx.as_fd(), 500) {
            continue;
        }
        let block = rx.recv_block().expect("recv block");
        for f in block.frames() {
            if f.data().windows(payload.len()).any(|w| w == payload) {
                assert!(f.data().len() >= frame.len());
                found = true;
                break 'outer;
            }
        }
    }
    assert!(found, "did not receive the injected frame on the ring");
}

#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn basic_sync_roundtrip() {
    let veth = VethPair::new("fnvb0", "fnvb1");
    let (mut tx, _) = Channel::builder(&veth.a).basic().build_sync().expect("open sender");
    let (_, mut rx) = Channel::builder(&veth.b).basic().build_sync().expect("open receiver");

    let payload = b"ferranet-basic-sync";
    let frame = make_frame(payload);
    tx.send(&frame).expect("send frame");

    let mut found = false;
    let deadline = std::time::Instant::now() + Duration::from_secs(3);
    'outer: while std::time::Instant::now() < deadline {
        if !wait_readable(rx.as_fd(), 500) {
            continue;
        }
        let block = rx.recv_block().expect("recv block");
        for f in block.frames() {
            if f.data().windows(payload.len()).any(|w| w == payload) {
                found = true;
                break 'outer;
            }
        }
    }
    assert!(found, "did not receive the injected frame (basic backend)");
}

#[cfg(feature = "tokio")]
#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn ring_async_roundtrip() {
    let veth = VethPair::new("fnva0", "fnva1");
    let rt = tokio::runtime::Builder::new_current_thread()
        .enable_all()
        .build()
        .expect("tokio runtime");

    rt.block_on(async {
        let (mut tx, _) = Channel::builder(&veth.a).build_async().expect("open sender");
        let (_, mut rx) =
            Channel::builder(&veth.b).rx_ring(rx_config()).build_async().expect("open receiver");

        let payload = b"ferranet-ring-async";
        let frame = make_frame(payload);
        tx.send(&frame).await.expect("send frame");

        let found = tokio::time::timeout(Duration::from_secs(3), async {
            loop {
                let block = rx.recv_block().await.expect("recv block");
                for f in block.frames() {
                    if f.data().windows(payload.len()).any(|w| w == payload) {
                        return true;
                    }
                }
            }
        })
        .await
        .unwrap_or(false);

        assert!(found, "did not receive the injected frame (async ring)");
    });
}

#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn fanout_distributes_across_receivers() {
    let veth = VethPair::new("fnvf0", "fnvf1");

    // Two receivers in one round-robin fanout group on B.
    let receivers = Channel::builder(&veth.b)
        .fanout(FanoutMode::LoadBalance)
        .rx_ring(rx_config())
        .build_fanout_rx(2)
        .expect("open fanout group");

    let stop = Arc::new(AtomicBool::new(false));

    // Count frames each receiver pulls until told to stop.
    let workers: Vec<_> = receivers
        .into_iter()
        .map(|mut rx| {
            let stop = stop.clone();
            thread::spawn(move || {
                let mut count: u64 = 0;
                while !stop.load(Ordering::Relaxed) {
                    if wait_readable(rx.as_fd(), 100) {
                        if let Ok(block) = rx.recv_block() {
                            count += block.frames().count() as u64;
                        }
                    }
                }
                count
            })
        })
        .collect();

    // Flood B's peer for a second.
    let (mut tx, _) = Channel::builder(&veth.a).build_sync().expect("open sender");
    let frames: Vec<Vec<u8>> = (0..64).map(|_| make_frame(b"fanout")).collect();
    let batch: Vec<&[u8]> = frames.iter().map(Vec::as_slice).collect();
    let deadline = Instant::now() + Duration::from_secs(1);
    while Instant::now() < deadline {
        let _ = tx.send_batch(&batch);
    }

    stop.store(true, Ordering::Relaxed);
    let counts: Vec<u64> = workers.into_iter().map(|w| w.join().unwrap()).collect();

    let total: u64 = counts.iter().sum();
    assert!(total > 0, "fanout group received no frames");
    // Round-robin should hand a meaningful share to *both* sockets, not just one.
    assert!(
        counts.iter().all(|&c| c > 0),
        "fanout did not distribute across both receivers: {counts:?}"
    );
}

#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn drop_detection_reports_losses() {
    let veth = VethPair::new("fnvd0", "fnvd1");

    // A tiny RX ring (one 4 KiB block) overflows almost immediately when not drained.
    let tiny = RingConfig { block_size: 4096, block_count: 4, retire_blk_tov_ms: 10, ..RingConfig::rx_default() };
    let (_tx, mut rx) = Channel::builder(&veth.b).rx_ring(tiny).build_sync().expect("open receiver");

    // Flood the peer hard from several threads so the receiver can't keep up.
    let stop = Arc::new(AtomicBool::new(false));
    let a = veth.a.clone();
    let flooders: Vec<_> = (0..4)
        .map(|_| {
            let stop = stop.clone();
            let a = a.clone();
            thread::spawn(move || {
                let (mut tx, _) = Channel::builder(&a).build_sync().expect("open sender");
                let frames: Vec<Vec<u8>> = (0..64).map(|_| make_frame(b"flood-flood-flood")).collect();
                let batch: Vec<&[u8]> = frames.iter().map(Vec::as_slice).collect();
                while !stop.load(Ordering::Relaxed) {
                    let _ = tx.send_batch(&batch);
                }
            })
        })
        .collect();

    // Drain under sustained overload: frames captured *after* a drop carry TP_STATUS_LOSING.
    // Don't call stats() in the loop — it would clear the flag before we observe it.
    let mut losing_seen = false;
    let deadline = Instant::now() + Duration::from_secs(2);
    while Instant::now() < deadline && !losing_seen {
        if wait_readable(rx.as_fd(), 200) {
            if let Ok(block) = rx.recv_block() {
                losing_seen |= block.is_losing();
            }
        }
    }
    let stats = rx.stats().expect("read stats");

    stop.store(true, Ordering::Relaxed);
    for f in flooders {
        let _ = f.join();
    }

    assert!(stats.dropped > 0, "expected kernel drops under overflow, got {stats:?}");
    assert!(losing_seen, "expected a block to carry the TP_STATUS_LOSING flag");
}

#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn default_fanout_groups_get_distinct_kernel_ids() {
    use std::os::fd::AsRawFd;

    let veth = VethPair::new("fnvg0", "fnvg1");

    // Two *independent* fanout groups with default ids on the same interface. Ids must be
    // kernel-allocated and distinct — unrelated captures sharing an id silently steal each
    // other's frames — and the second socket of each group must join the first's id.
    let g1 = Channel::builder(&veth.b).rx_ring(rx_config()).build_fanout_rx(2).expect("group 1");
    let g2 = Channel::builder(&veth.b).rx_ring(rx_config()).build_fanout_rx(2).expect("group 2");

    let group_id = |rx: &ferranet::Receiver| -> u16 {
        let mut val: libc::c_int = 0;
        let mut len = std::mem::size_of::<libc::c_int>() as libc::socklen_t;
        // SAFETY: `val`/`len` are valid storage and its length for getsockopt to fill.
        let rc = unsafe {
            libc::getsockopt(
                rx.as_fd().as_raw_fd(),
                libc::SOL_PACKET,
                libc::PACKET_FANOUT,
                (&raw mut val).cast(),
                &raw mut len,
            )
        };
        assert_eq!(rc, 0, "read PACKET_FANOUT back");
        (val & 0xffff) as u16
    };

    assert_eq!(group_id(&g1[0]), group_id(&g1[1]), "sockets of one group share its id");
    assert_eq!(group_id(&g2[0]), group_id(&g2[1]), "sockets of one group share its id");
    assert_ne!(
        group_id(&g1[0]),
        group_id(&g2[0]),
        "independent default-id groups must not share a fanout group"
    );
}

#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn tx_ring_recovers_after_kernel_rejects_frame() {
    let veth = VethPair::new("fnvw0", "fnvw1");
    // Lower the MTU so a frame can pass the ring's local size check (frame_size 2048) yet be
    // rejected by the kernel, which hands the slot back marked TP_STATUS_WRONG_FORMAT.
    assert!(run("ip", &["link", "set", &veth.a, "mtu", "1000"]), "failed to lower MTU");

    // A two-slot TX ring, so the cursor quickly wraps back onto the rejected slot.
    let tiny_tx = RingConfig { block_size: 4096, block_count: 1, ..RingConfig::tx_default() };
    let (mut tx, _) =
        Channel::builder(&veth.a).tx_ring(tiny_tx).build_sync().expect("open sender");

    let oversized = make_frame(&[0u8; 1400]);
    assert!(tx.send(&oversized).is_err(), "kernel should reject an over-MTU frame");

    // The ring must stay usable afterwards — before reclaiming WRONG_FORMAT slots, the wrap
    // onto the rejected slot returned WouldBlock forever.
    for i in 0..8 {
        let frame = make_frame(format!("recover-{i}").as_bytes());
        tx.send(&frame).unwrap_or_else(|e| panic!("send {i} after rejection failed: {e}"));
    }
}

#[test]
#[ignore = "requires CAP_NET_RAW/CAP_NET_ADMIN; run inside a network namespace"]
fn batched_send_counts() {
    let veth = VethPair::new("fnvt0", "fnvt1");
    let (mut tx, _) = Channel::builder(&veth.a).build_sync().expect("open sender");

    let f1 = make_frame(b"batch-1");
    let f2 = make_frame(b"batch-2");
    let f3 = make_frame(b"batch-3");
    let batch: [&[u8]; 3] = [&f1, &f2, &f3];
    let sent = tx.send_batch(&batch).expect("send batch");
    assert_eq!(sent, 3, "all three frames should be accepted by the TX ring");

    // An oversized frame in a batch must be reported the same way `send` reports it.
    let oversized = make_frame(&[0u8; 3000]); // > the 2 KiB TX frame slots
    let bad_batch: [&[u8]; 2] = [&f1, &oversized];
    let err = tx.send_batch(&bad_batch).expect_err("oversized frame in batch");
    assert!(
        matches!(err, ferranet::Error::FrameTooLarge { .. }),
        "expected FrameTooLarge, got {err:?}"
    );
}