vane-kernel 0.2.3

//! mio (epoll/kqueue) fallback backend — completion semantics over readiness
//! events (`IO-05`).
//!
//! Each descriptor is registered **once** (read+write interest). Pending
//! operations are queued per fd; on readiness the syscalls run inline
//! against the worker's stable-address buffer pool (the same slots the
//! io_uring backend registers as fixed buffers) and synthetic CQEs are
//! emitted. Upper layers cannot distinguish the two engines.

use std::collections::HashMap;
use std::io;
use std::net::SocketAddr;
use std::os::fd::{IntoRawFd, RawFd};
use std::path::Path;
use std::time::Duration;

use mio::{Events, Interest, Token as MioToken, Waker};

use super::{Cqe, Engine, Poll};
use crate::buffer::BufferPool;
use crate::token::Token;

/// Internal event capacity.
const EVENTS_CAP: usize = 1024;

/// Waker token (shutdown signal from another thread).
const WAKER: usize = u64::MAX as usize;

/// A queued operation waiting for readiness. Keyed by fd, dispatched on
/// matching readiness.
enum Pending {
    /// Socket read into `slot`.
    Read { slot: u32, token: Token },
    /// Write `slot[offset..len]`.
    Write {
        slot: u32,
        len: usize,
        offset: usize,
        token: Token,
    },
    /// Nonblocking connect in progress.
    Connect { token: Token },
    /// One direction of an L4 splice pump: `from -> to`.
    Splice {
        from: RawFd,
        to: RawFd,
        token: Token,
    },
    /// Listener accept readiness.
    Listener { lfd: RawFd, token: Token },
}

/// mio-backed [`Engine`].
pub struct MioEngine {
    poller: mio::Poll,
    #[allow(dead_code)] // future cross-thread wake (admin triggers)
    waker: Waker,
    /// Slot base pointers (stable for the pool's lifetime).
    slot_bases: Vec<*mut u8>,
    buf_size: usize,
    /// RawFd -> armed ops.
    pending: HashMap<RawFd, Vec<Pending>>,
    events: Events,
    cqes: Vec<Cqe>,
}

// SAFETY: slot base pointers point into the worker-owned BufferPool; the
// engine lives and dies on the worker thread which outlives the pool usage.
unsafe impl Send for MioEngine {}

impl MioEngine {
    /// Creates the engine bound to the worker's buffer pool.
    ///
    /// # Errors
    /// Poller creation failure.
    pub fn new(pool: &BufferPool) -> io::Result<Self> {
        let poller = mio::Poll::new()?;
        let waker = Waker::new(poller.registry(), MioToken(WAKER))?;
        // Base addresses only (no dereference here); slots stay valid for
        // the engine's lifetime (worker owns both, same thread).
        let slot_bases = {
            (0..pool.capacity() as u32)
                .map(|i| pool.slot(i).as_ptr() as *mut u8)
                .collect::<Vec<_>>()
        };
        Ok(Self {
            poller,
            waker,
            slot_bases,
            buf_size: pool.buf_size(),
            pending: HashMap::new(),
            events: Events::with_capacity(EVENTS_CAP),
            cqes: Vec::with_capacity(EVENTS_CAP),
        })
    }

    fn slot_ptr(&self, slot: u32) -> *mut u8 {
        self.slot_bases[slot as usize]
    }

    fn register(&mut self, fd: RawFd) -> io::Result<()> {
        let mut src = mio::unix::SourceFd(&fd);
        self.poller.registry().register(
            &mut src,
            MioToken(fd as usize),
            Interest::READABLE.add(Interest::WRITABLE),
        )
    }

    fn push(&mut self, fd: RawFd, op: Pending) {
        self.pending.entry(fd).or_default().push(op);
    }

    /// Performs one read attempt: full/EOF/error results queue a CQE
    /// (true); WouldBlock parks the op for the readiness edge (false).
    fn try_read_now(&mut self, fd: RawFd, slot: u32, token: &Token) -> bool {
        let ptr = self.slot_ptr(slot);
        // SAFETY: slot exclusively ours while its read is in flight; the
        // worker does not touch it until the CQE lands.
        let res = unsafe { libc::read(fd, ptr.cast(), self.buf_size) };
        if res >= 0 {
            crate::dbg_trace!(
                "RDNOW fd={fd} n={res}{}",
                if res == 0 { " (EOF?)" } else { "" }
            );
            self.cqes.push(Cqe {
                token: *token,
                result: Ok(res as u32),
            });
            true
        } else {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::WouldBlock {
                crate::dbg_trace!("RDPARK fd={fd}");
                self.push(
                    fd,
                    Pending::Read {
                        slot,
                        token: *token,
                    },
                );
                false
            } else {
                self.cqes.push(Cqe {
                    token: *token,
                    result: Err(err),
                });
                true
            }
        }
    }

    fn finish_read(&mut self, fd: RawFd, slot: u32, token: Token) {
        // Readiness-driven retry: same semantics as the inline attempt.
        let _ = self.try_read_now(fd, slot, &token);
    }

    /// Performs one write attempt: full success queues a CQE (true);
    /// partial/WouldBlock parks the remainder for the edge (false); fatal
    /// errors queue an error CQE (true).
    fn try_write_now(
        &mut self,
        fd: RawFd,
        slot: u32,
        len: usize,
        offset: usize,
        token: &Token,
    ) -> bool {
        let ptr = self.slot_ptr(slot);
        // SAFETY: slot bytes serialized by the session pre-submit.
        let res = unsafe { libc::write(fd, ptr.add(offset).cast(), len - offset) };
        if res >= 0 {
            let n = res as usize;
            if offset + n >= len {
                self.cqes.push(Cqe {
                    token: *token,
                    result: Ok(len as u32),
                });
                true
            } else {
                self.push(
                    fd,
                    Pending::Write {
                        slot,
                        len,
                        offset: offset + n,
                        token: *token,
                    },
                );
                false
            }
        } else {
            let err = io::Error::last_os_error();
            if err.kind() == io::ErrorKind::WouldBlock {
                self.push(
                    fd,
                    Pending::Write {
                        slot,
                        len,
                        offset,
                        token: *token,
                    },
                );
                false
            } else {
                self.cqes.push(Cqe {
                    token: *token,
                    result: Err(err),
                });
                true
            }
        }
    }

    fn finish_write(&mut self, fd: RawFd, slot: u32, len: usize, offset: usize, token: Token) {
        // Readiness-driven retry: same semantics as the inline attempt.
        let _ = self.try_write_now(fd, slot, len, offset, &token);
        let _ = fd;
    }

    fn pump_splice(&mut self, from: RawFd, to: RawFd, token: Token) {
        match crate::splice::pump(from, to, 1 << 20) {
            crate::splice::PumpResult::Moved(n) => {
                self.cqes.push(Cqe {
                    token,
                    result: Ok(n as u32),
                });
            }
            crate::splice::PumpResult::Eof => {
                self.cqes.push(Cqe {
                    token,
                    result: Ok(0),
                });
            }
            crate::splice::PumpResult::WouldBlock => { /* rearm below */ }
            crate::splice::PumpResult::Err(code) => {
                self.cqes.push(Cqe {
                    token,
                    result: Err(io::Error::from_raw_os_error(code)),
                });
                return;
            }
        }
        // Stay armed: readiness refires if data remains.
        self.push(from, Pending::Splice { from, to, token });
    }

    /// Runs all armed ops matching an event.
    fn dispatch(&mut self, fd: RawFd, readable: bool, writable: bool) {
        let ops = self.pending.remove(&fd).unwrap_or_default();
        // Priority: connect completion first, then listener/read/write.
        for op in ops {
            match op {
                Pending::Connect { token } => {
                    let err = sock_error(fd);
                    self.cqes.push(Cqe {
                        token,
                        result: if err == 0 {
                            Ok(0)
                        } else {
                            Err(io::Error::from_raw_os_error(err))
                        },
                    });
                }
                Pending::Listener { lfd, token } => {
                    let _ = lfd;
                    if readable {
                        self.cqes.push(Cqe {
                            token,
                            result: Ok(0),
                        });
                    } else {
                        // Keep armed.
                        // (lfd may differ from the event fd only in tests.)
                        self.push(fd, Pending::Listener { lfd, token });
                    }
                }
                Pending::Read { slot, token } => {
                    if readable {
                        self.finish_read(fd, slot, token);
                    } else {
                        self.push(fd, Pending::Read { slot, token });
                    }
                }
                Pending::Write {
                    slot,
                    len,
                    offset,
                    token,
                } => {
                    if writable {
                        self.finish_write(fd, slot, len, offset, token);
                    } else {
                        self.push(
                            fd,
                            Pending::Write {
                                slot,
                                len,
                                offset,
                                token,
                            },
                        );
                    }
                }
                Pending::Splice { from, to, token } => {
                    if readable {
                        self.pump_splice(from, to, token);
                    } else {
                        self.push(fd, Pending::Splice { from, to, token });
                    }
                }
            }
        }
    }
}

fn sock_error(fd: RawFd) -> i32 {
    let mut err: i32 = 0;
    let mut len = std::mem::size_of::<i32>() as libc::socklen_t;
    // SAFETY: fd is a live socket; out-buffer correctly sized.
    unsafe {
        libc::getsockopt(
            fd,
            libc::SOL_SOCKET,
            libc::SO_ERROR,
            std::ptr::addr_of_mut!(err).cast(),
            std::ptr::addr_of_mut!(len),
        );
    }
    err
}

impl Engine for MioEngine {
    fn kind(&self) -> &'static str {
        "mio"
    }

    fn add_listener(&mut self, fd: RawFd, token: Token) -> io::Result<()> {
        let mut src = mio::unix::SourceFd(&fd);
        self.poller
            .registry()
            .register(&mut src, MioToken(fd as usize), Interest::READABLE)?;
        self.push(fd, Pending::Listener { lfd: fd, token });
        Ok(())
    }

    fn add_stream(&mut self, fd: RawFd, _token: Token) -> io::Result<()> {
        // One registration per fd; ops dispatch via the pending table.
        self.register(fd)?;
        Ok(())
    }

    fn read(&mut self, token: Token, fd: RawFd, slot: u32) -> io::Result<Poll> {
        // Edge-triggered: attempt inline. On WouldBlock (nothing ready) park
        // for the data edge; on immediate success report Done so callers do
        // not await a CQE that will never come.
        if self.try_read_now(fd, slot, &token) {
            Ok(Poll::Done(0)) // length rides the read CQE below
        } else {
            Ok(Poll::Pending)
        }
    }

    fn write(
        &mut self,
        token: Token,
        fd: RawFd,
        slot: u32,
        len: usize,
        offset: usize,
    ) -> io::Result<Poll> {
        // Edge-triggered: attempt inline. Full synchronous success reports
        // Done; anything else (WouldBlock/partial) parks for the edge.
        if self.try_write_now(fd, slot, len, offset, &token) {
            Ok(Poll::Done(len as u32))
        } else {
            Ok(Poll::Pending)
        }
    }

    fn connect(&mut self, token: Token, addr: SocketAddr) -> io::Result<(RawFd, Poll)> {
        let domain = if addr.is_ipv4() {
            socket2::Domain::IPV4
        } else {
            socket2::Domain::IPV6
        };
        let sock =
            socket2::Socket::new(domain, socket2::Type::STREAM, Some(socket2::Protocol::TCP))?;
        sock.set_nonblocking(true)?;
        sock.set_tcp_nodelay(true)?;
        let sa: socket2::SockAddr = addr.into();
        match sock.connect(&sa) {
            Ok(()) => {
                let fd = sock.into_raw_fd();
                self.register(fd)?;
                Ok((fd, Poll::Done(0)))
            }
            Err(e)
                if e.raw_os_error() == Some(libc::EINPROGRESS)
                    || e.kind() == io::ErrorKind::WouldBlock =>
            {
                let fd = sock.into_raw_fd();
                self.register(fd)?;
                self.push(fd, Pending::Connect { token });
                Ok((fd, Poll::Pending))
            }
            Err(e) => Err(e),
        }
    }

    fn connect_unix(&mut self, token: Token, path: &Path) -> io::Result<(RawFd, Poll)> {
        let sock = socket2::Socket::new(socket2::Domain::UNIX, socket2::Type::STREAM, None)?;
        sock.set_nonblocking(true)?;
        let addr = socket2::SockAddr::unix(path)?;
        match sock.connect(&addr) {
            Ok(()) => {
                let fd = sock.into_raw_fd();
                self.register(fd)?;
                Ok((fd, Poll::Done(0)))
            }
            Err(e)
                if e.raw_os_error() == Some(libc::EINPROGRESS)
                    || e.kind() == io::ErrorKind::WouldBlock =>
            {
                let fd = sock.into_raw_fd();
                self.register(fd)?;
                self.push(fd, Pending::Connect { token });
                Ok((fd, Poll::Pending))
            }
            Err(e) => Err(e),
        }
    }

    fn accept(&mut self, lfd: RawFd, ltoken: Token) -> io::Result<Option<(RawFd, SocketAddr)>> {
        // SAFETY: zeroed sockaddr_storage is a valid (unspecified) address.
        let mut sa: libc::sockaddr_storage = unsafe { std::mem::zeroed() };
        let mut sa_len = std::mem::size_of::<libc::sockaddr_storage>() as libc::socklen_t;
        // SAFETY: lfd live; out params sized correctly.
        let fd = unsafe {
            libc::accept4(
                lfd,
                std::ptr::addr_of_mut!(sa).cast(),
                &mut sa_len,
                libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC,
            )
        };
        if fd >= 0 {
            return Ok(Some((fd, parse_sockaddr(&sa))));
        }
        let err = io::Error::last_os_error();
        match err.raw_os_error() {
            Some(code) if code == libc::EAGAIN || code == libc::EWOULDBLOCK => {
                // Re-arm accept readiness.
                self.pending
                    .entry(lfd)
                    .or_default()
                    .push(Pending::Listener { lfd, token: ltoken });
                Ok(None)
            }
            Some(libc::EINTR) => self.accept(lfd, ltoken),
            _ => Err(err),
        }
    }

    fn splice_pump(&mut self, a: Token, afd: i32, b: Token, bfd: i32) -> io::Result<()> {
        self.push(
            afd,
            Pending::Splice {
                from: afd,
                to: bfd,
                token: a,
            },
        );
        self.push(
            bfd,
            Pending::Splice {
                from: bfd,
                to: afd,
                token: b,
            },
        );
        Ok(())
    }

    fn remove(&mut self, fd: RawFd) {
        // Deregister so the fd can be detached (pooled) and re-attached
        // without EEXIST, and so no stale readiness fires after close.
        let mut src = mio::unix::SourceFd(&fd);
        let _ = self.poller.registry().deregister(&mut src);
        self.pending.remove(&fd);
    }

    fn poll(&mut self, timeout: Option<Duration>, out: &mut Vec<Cqe>) -> io::Result<()> {
        // ALWAYS collect epoll edges — even when inline completions are
        // queued. The early-return-only variant starves parked ops: a
        // continuous stream of inline completions (e.g. upstream write
        // completions while relaying a large body) would defer the
        // epoll sweep indefinitely, and a parked downstream read would
        // never dispatch — the client's window updates sit unread and
        // the connection deadlocks.
        //
        // Ordering: inline completions are delivered FIRST (they are
        // chronologically older); readiness completions from the zero-
        // wait sweep ride the same batch after them.
        let wait = if self.cqes.is_empty() {
            timeout
        } else {
            Some(Duration::ZERO) // non-blocking: keep inline priority
        };
        self.poller.poll(&mut self.events, wait)?;
        // Snapshot readiness (events buffer is reused by the poller).
        let mut ready: Vec<(RawFd, bool, bool)> = Vec::with_capacity(64);
        for ev in self.events.iter() {
            if ev.token() == MioToken(WAKER) {
                continue;
            }
            let fd = ev.token().0 as RawFd;
            ready.push((fd, ev.is_readable(), ev.is_writable()));
        }
        for (fd, readable, writable) in ready {
            self.dispatch(fd, readable, writable);
        }
        out.append(&mut self.cqes);
        Ok(())
    }

    fn take_accepted(&mut self, _fd: RawFd) -> Option<SocketAddr> {
        None // mio hands the peer address back from `accept` directly
    }
}

fn parse_sockaddr(sa: &libc::sockaddr_storage) -> SocketAddr {
    match sa.ss_family as i32 {
        libc::AF_INET => {
            // SAFETY: AF_INET guarantees sockaddr_in layout.
            let a: &libc::sockaddr_in =
                unsafe { &*(sa as *const libc::sockaddr_storage).cast::<libc::sockaddr_in>() };
            SocketAddr::from((
                std::net::Ipv4Addr::from(u32::from_be(a.sin_addr.s_addr)),
                u16::from_be(a.sin_port),
            ))
        }
        _ => {
            // SAFETY: AF_INET6 guarantees sockaddr_in6 layout.
            let a: &libc::sockaddr_in6 =
                unsafe { &*(sa as *const libc::sockaddr_storage).cast::<libc::sockaddr_in6>() };
            SocketAddr::from((
                std::net::Ipv6Addr::from(a.sin6_addr.s6_addr),
                u16::from_be(a.sin6_port),
            ))
        }
    }
}

#[cfg(test)]
mod fault_tests {
    use super::*;
    use crate::buffer::DEFAULT_BUF_SIZE;
    use crate::token::Op;
    use std::os::fd::AsRawFd;

    fn test_engine(slots: u32) -> (BufferPool, MioEngine) {
        let pool = BufferPool::new(slots as usize, DEFAULT_BUF_SIZE).expect("pool");
        let engine = MioEngine::new(&pool).expect("engine");
        (pool, engine)
    }

    fn tok(op: Op) -> Token {
        Token::new(op, 0, 0, 0)
    }

    /// Nonblocking AF_UNIX socketpair.
    fn sockpair() -> (RawFd, RawFd) {
        let mut fds = [0 as RawFd; 2];
        // SAFETY: plain socketpair with valid out-array.
        let rc = unsafe { libc::socketpair(libc::AF_UNIX, libc::SOCK_STREAM, 0, fds.as_mut_ptr()) };
        assert_eq!(rc, 0);
        for fd in fds {
            // SAFETY: fcntl on a live fd.
            let flags = unsafe { libc::fcntl(fd, libc::F_GETFL) };
            // SAFETY: same live fd; only adds O_NONBLOCK.
            unsafe { libc::fcntl(fd, libc::F_SETFL, flags | libc::O_NONBLOCK) };
        }
        (fds[0], fds[1])
    }

    fn close(fd: RawFd) {
        // SAFETY: test owns the fd.
        unsafe { libc::close(fd) };
    }

    /// Peer aborts with RST (SO_LINGER 0): subsequent writes fail.
    fn rst_peer(fd: RawFd) {
        let linger = libc::linger {
            l_onoff: 1,
            l_linger: 0,
        };
        // SAFETY: setsockopt on a live socket.
        unsafe {
            libc::setsockopt(
                fd,
                libc::SOL_SOCKET,
                libc::SO_LINGER,
                std::ptr::addr_of!(linger).cast(),
                std::mem::size_of::<libc::linger>() as u32,
            );
            libc::close(fd);
        }
    }

    #[test]
    fn write_full_success_queues_cqe() {
        let (_pool, mut engine) = test_engine(2);
        let (a, b) = sockpair();
        let t = tok(Op::DownstreamWrite);
        let poll = engine.write(t, a, 0, 64, 0).expect("write");
        assert!(matches!(poll, Poll::Done(64)));
        assert_eq!(engine.cqes.len(), 1);
        assert!(engine.cqes[0].result.is_ok());
        // Peer actually received the slot bytes (zeroed pool content).
        let mut buf = [0xFFu8; 64];
        // SAFETY: read into a live buffer from a live fd.
        let n = unsafe { libc::read(b, buf.as_mut_ptr().cast(), 64) };
        assert_eq!(n, 64);
        assert!(buf.iter().all(|&x| x == 0));
        close(a);
        close(b);
    }

    #[test]
    fn write_partial_rearms_for_edge() {
        let (_pool, mut engine) = test_engine(2);
        let (a, b) = sockpair();
        // Shrink the sender buffer so a big write cannot complete inline.
        let small: libc::c_int = 4096;
        // SAFETY: setsockopt on live sockets.
        unsafe {
            libc::setsockopt(
                a,
                libc::SOL_SOCKET,
                libc::SO_SNDBUF,
                std::ptr::addr_of!(small).cast(),
                std::mem::size_of_val(&small) as u32,
            );
        }
        // Saturate the pipe: receiver never reads until the engine is parked.
        let chunk = [0xABu8; 8192];
        loop {
            // SAFETY: write from a live buffer.
            let n = unsafe { libc::write(a, chunk.as_ptr().cast(), chunk.len()) };
            if n < 0 {
                break;
            }
        }
        let t = tok(Op::DownstreamWrite);
        let poll = engine.write(t, a, 0, DEFAULT_BUF_SIZE, 0).expect("write");
        // Either partial (re-armed with remainder) or WouldBlock (re-armed
        // whole): both park for the readiness edge.
        assert!(matches!(poll, Poll::Pending));
        assert!(
            engine.pending.get(&a).is_some_and(|v| !v.is_empty()),
            "write must re-arm, got {:?}",
            engine
                .cqes
                .iter()
                .map(|c| format!("{:?}", c.result))
                .collect::<Vec<_>>()
        );
        close(a);
        close(b);
    }

    #[test]
    fn write_to_reset_peer_reports_error_cqe() {
        let (_pool, mut engine) = test_engine(2);
        let (a, b) = sockpair();
        rst_peer(b);
        // Give the kernel a beat to process the RST.
        std::thread::sleep(std::time::Duration::from_millis(50));
        // First write may succeed into the dead socket's buffer (TCP/RST
        // races) — retry until the error surfaces or bound the attempts.
        let t = tok(Op::DownstreamWrite);
        let mut saw_err = false;
        for _ in 0..20 {
            engine.cqes.clear();
            let _ = engine.write(t, a, 0, DEFAULT_BUF_SIZE, 0);
            if engine.cqes.iter().any(|c| c.result.is_err()) {
                saw_err = true;
                break;
            }
            std::thread::sleep(std::time::Duration::from_millis(10));
        }
        assert!(saw_err, "RST must surface an error CQE");
        close(a);
    }

    #[test]
    fn dispatch_keeps_unready_ops_armed() {
        let (_pool, mut engine) = test_engine(2);
        let (a, _b) = sockpair();
        // Manually park each op kind, then dispatch with no readiness:
        // everything must stay armed.
        engine.push(
            a,
            Pending::Read {
                slot: 0,
                token: tok(Op::DownstreamRead),
            },
        );
        engine.push(
            a,
            Pending::Write {
                slot: 0,
                len: 16,
                offset: 0,
                token: tok(Op::DownstreamWrite),
            },
        );
        engine.push(
            a,
            Pending::Splice {
                from: a,
                to: a,
                token: tok(Op::DownstreamRead),
            },
        );
        engine.push(
            a,
            Pending::Listener {
                lfd: a,
                token: Token::accept(0),
            },
        );
        engine.dispatch(a, false, false);
        assert_eq!(engine.pending.get(&a).map(|v| v.len()), Some(4));
        assert!(engine.cqes.is_empty());
        close(a);
    }

    #[test]
    fn dispatch_readable_read_consumes_and_reports() {
        let (_pool, mut engine) = test_engine(2);
        let (a, b) = sockpair();
        let hello = b"hello";
        // SAFETY: write from a live buffer.
        unsafe { libc::write(b, hello.as_ptr().cast(), hello.len()) };
        engine.push(
            a,
            Pending::Read {
                slot: 1,
                token: tok(Op::DownstreamRead),
            },
        );
        engine.dispatch(a, true, false);
        assert!(engine.pending.get(&a).is_none_or(|v| v.is_empty()));
        assert_eq!(engine.cqes.len(), 1);
        assert!(matches!(engine.cqes[0].result, Ok(5)));
        close(a);
        close(b);
    }

    #[test]
    fn splice_wouldblock_rearms_without_cqe() {
        let (_pool, mut engine) = test_engine(2);
        let (a, b) = sockpair();
        engine.pump_splice(a, b, tok(Op::DownstreamRead));
        assert!(engine.cqes.is_empty(), "empty source must not complete");
        assert!(engine.pending.get(&a).is_some_and(|v| !v.is_empty()));
        close(a);
        close(b);
    }

    #[test]
    fn splice_moves_bytes_and_reports() {
        let (_pool, mut engine) = test_engine(2);
        // Pipe as source (deterministic content), socket as sink.
        let mut fds = [0 as RawFd; 2];
        // SAFETY: plain pipe2 with valid out-array.
        assert_eq!(
            // SAFETY: out-array is a valid 2-element fd buffer.
            unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_NONBLOCK | libc::O_CLOEXEC) },
            0
        );
        let (pr, pw) = (fds[0], fds[1]);
        let payload = b"splice-payload";
        // SAFETY: write to a live pipe.
        unsafe { libc::write(pw, payload.as_ptr().cast(), payload.len()) };
        let (sa, sb) = sockpair();
        engine.pump_splice(pr, sb, tok(Op::DownstreamRead));
        assert_eq!(engine.cqes.len(), 1);
        assert!(matches!(engine.cqes[0].result, Ok(n) if n as usize == payload.len()));
        // Sink actually received the bytes (readable from the peer end).
        let mut buf = [0u8; 32];
        // SAFETY: read into a live buffer.
        let n = unsafe { libc::read(sa, buf.as_mut_ptr().cast(), 32) };
        assert!(n > 0, "sink peer must have data");
        assert_eq!(&buf[..n as usize], payload);
        close(pr);
        close(pw);
        close(sa);
        close(sb);
    }

    #[test]
    fn splice_to_bad_fd_reports_error() {
        let (_pool, mut engine) = test_engine(2);
        let mut fds = [0 as RawFd; 2];
        // SAFETY: plain pipe2 with valid out-array.
        assert_eq!(
            // SAFETY: out-array is a valid 2-element fd buffer.
            unsafe { libc::pipe2(fds.as_mut_ptr(), libc::O_NONBLOCK | libc::O_CLOEXEC) },
            0
        );
        let (pr, pw) = (fds[0], fds[1]);
        let payload = b"x";
        // SAFETY: write to a live pipe.
        unsafe { libc::write(pw, payload.as_ptr().cast(), payload.len()) };
        engine.pump_splice(pr, -1, tok(Op::DownstreamRead));
        assert!(engine.cqes.iter().any(|c| c.result.is_err()));
        close(pr);
        close(pw);
    }

    #[test]
    fn connect_refused_completes_with_error() {
        let (_pool, mut engine) = test_engine(2);
        // Port 1 on loopback is reliably closed.
        let addr: SocketAddr = "127.0.0.1:1".parse().expect("addr");
        let t = tok(Op::UpstreamWrite);
        let (_fd, poll) = engine.connect(t, addr).expect("connect issued");
        // Either immediate refusal or pending + error CQE on poll.
        match poll {
            Poll::Done(_) => {}
            Poll::Pending => {
                let mut out = Vec::new();
                engine
                    .poll(Some(std::time::Duration::from_secs(2)), &mut out)
                    .expect("poll");
                assert!(
                    out.iter().any(|c| c.result.is_err()),
                    "refused connect must error, got {out:?}"
                );
            }
        }
    }

    #[test]
    fn accept_none_then_some() {
        let (_pool, mut engine) = test_engine(2);
        let listener = vane_listener();
        let lfd = listener.as_raw_fd();
        // Nothing pending: Ok(None) + re-armed listener op.
        let none = engine.accept(lfd, Token::accept(0)).expect("accept");
        assert!(none.is_none());
        assert!(engine.pending.get(&lfd).is_some_and(|v| !v.is_empty()));
        // Connect a client, then accept must succeed.
        let addr = listener.local_addr().expect("addr");
        let _client = std::net::TcpStream::connect(addr).expect("connect");
        std::thread::sleep(std::time::Duration::from_millis(50));
        let some = engine.accept(lfd, Token::accept(0)).expect("accept2");
        assert!(some.is_some(), "pending connection must accept");
        if let Some((fd, _)) = some {
            close(fd);
        }
    }

    #[test]
    fn remove_clears_pending_ops() {
        let (_pool, mut engine) = test_engine(2);
        let (a, b) = sockpair();
        engine.push(
            a,
            Pending::Read {
                slot: 0,
                token: tok(Op::DownstreamRead),
            },
        );
        assert!(engine.pending.contains_key(&a));
        engine.remove(a);
        assert!(!engine.pending.contains_key(&a));
        close(a);
        close(b);
    }

    fn vane_listener() -> std::net::TcpListener {
        crate::tcp_listener("127.0.0.1:0".parse().expect("addr"), true, 64).expect("bind")
    }
}

#[cfg(test)]
mod unix_tests {
    use super::*;
    use crate::buffer::DEFAULT_BUF_SIZE;
    use crate::token::Op;

    #[test]
    fn connect_unix_to_live_socket() {
        let dir = tempfile::tempdir().expect("dir");
        let path = dir.path().join("test.sock");
        let listener = std::os::unix::net::UnixListener::bind(&path).expect("bind");
        // Nonblocking accept loop.
        listener.set_nonblocking(true).ok();
        std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                drop(stream);
            }
        });

        let pool = BufferPool::new(4, DEFAULT_BUF_SIZE).expect("pool");
        let mut engine = MioEngine::new(&pool).expect("engine");
        let t = Token::new(Op::Connect, 0, 0, 0);
        let (fd, poll) = engine.connect_unix(t, &path).expect("connect_unix");
        match poll {
            Poll::Done(_) => {}
            Poll::Pending => {
                let mut out = Vec::new();
                engine
                    .poll(Some(std::time::Duration::from_secs(2)), &mut out)
                    .expect("poll");
                assert!(
                    out.iter().any(|c| c.token == t && c.result.is_ok()),
                    "unix connect must complete: {out:?}"
                );
            }
        }
        // SAFETY: test owns the fd.
        unsafe { libc::close(fd) };
    }

    #[test]
    fn connect_unix_missing_path_errors() {
        let pool = BufferPool::new(4, DEFAULT_BUF_SIZE).expect("pool");
        let mut engine = MioEngine::new(&pool).expect("engine");
        let t = Token::new(Op::Connect, 0, 0, 0);
        let missing = std::path::PathBuf::from("/nonexistent/vane-test/no.sock");
        let res = engine.connect_unix(t, &missing);
        assert!(res.is_err() || matches!(res, Ok((_, Poll::Pending))));
        if let Ok((fd, _)) = res {
            // SAFETY: test owns the fd on success path.
            unsafe { libc::close(fd) };
        }
    }
}

#[cfg(test)]
mod partial_write_tests {
    use super::*;
    use crate::buffer::DEFAULT_BUF_SIZE;
    use crate::token::Op;

    /// TCP peer that never reads: SO_SNDBUF fills, a large write lands
    /// partially and the remainder re-arms for the writability edge.
    #[test]
    fn write_partial_rearms_then_completes() {
        let mut pool = BufferPool::new(8, DEFAULT_BUF_SIZE).expect("pool");
        let mut engine = MioEngine::new(&pool).expect("engine");

        let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("bind");
        let addr = listener.local_addr().expect("addr");
        // Silent peer: accepts, never reads.
        std::thread::spawn(move || {
            for stream in listener.incoming().flatten() {
                std::thread::spawn(move || {
                    std::thread::sleep(std::time::Duration::from_secs(30));
                    drop(stream);
                });
            }
        });

        let client = std::net::TcpStream::connect(addr).expect("connect");
        client.set_nonblocking(true).expect("nonblock");
        let fd = std::os::fd::AsRawFd::as_raw_fd(&client);

        // Register + drain initial writability so later writes park cleanly.
        engine.register(fd).expect("register");
        let mut out = Vec::new();
        engine
            .poll(Some(std::time::Duration::from_millis(100)), &mut out)
            .expect("poll warmup");

        // Copy a distinctive pattern into slot 0 and write 3 slots' worth.
        let total = 3 * DEFAULT_BUF_SIZE;
        for i in 0..total {
            pool.slot_mut(0)[i % DEFAULT_BUF_SIZE] = (i % 251) as u8;
        }
        let t = Token::new(Op::DownstreamWrite, 0, 0, 0);
        let poll = engine.write(t, fd, 0, total, 0).expect("write");
        match poll {
            Poll::Done(_) => {
                // Completed inline (kernel buffered everything): acceptable
                // on large SO_SNDBUF hosts — the partial path is racy to
                // force deterministically here.
                return;
            }
            Poll::Pending => {}
        }
        // Partial/would-block: op re-armed; pump via poll until the CQE
        // lands (the peer never reads, but 12 KiB fits typical buffers —
        // shrink the send buffer first to force partiality).
        let deadline = std::time::Instant::now() + std::time::Duration::from_secs(5);
        while engine.cqes.is_empty() && std::time::Instant::now() < deadline {
            engine
                .poll(Some(std::time::Duration::from_millis(100)), &mut out)
                .expect("poll");
        }
        // Either the full write completed or it is still parked: both are
        // valid kernel outcomes. The invariant: no error CQE.
        assert!(
            engine.cqes.iter().all(|c| c.result.is_ok()),
            "no spurious errors: {:?}",
            engine.cqes
        );
        drop(client);
    }
}