net-mumu 0.2.0-rc.3

// src/real_ping.rs
//
// Raw ICMP (Echo) ping — single-socket, **pull-driven** transform (via iterator adapter).
// No shells, no timers, no sleeps. One probe per pull.
// Let your Flow graph govern cadence (flow:throttle) and lifetime (flow:stop_when).
//
// Public API (used by src/bridge.rs):
//   - RealPingOptions
//   - spawn_iterator(opts: RealPingOptions) -> IteratorHandle
//   - iter_to_transform(handle: IteratorHandle) -> Box<FunctionValue>
//
// Rows emitted (KeyedArray):
//   { dest, ip, seq, ok, bytes?, ttl?, ms? }            // on reply
//   { dest, ip: "0.0.0.0", seq: 0, ok:false, message }  // single-setup error row then EOF
//
// Linux first: prefer ICMP "ping socket" (SOCK_DGRAM + IPPROTO_ICMP + ICMP_FILTER),
// fallback to RAW ICMP (SOCK_RAW + IPPROTO_ICMP). Other Unix: RAW. Non-Unix: stub.

use indexmap::IndexMap;
use mumu::parser::types::{FunctionValue, IteratorHandle, IteratorKind, PluginIterator, Value};
use std::net::IpAddr;
use std::sync::{Arc, Mutex};

/// Options are accepted for compatibility. The pull rate and lifetime
/// should be governed by your Flow pipeline. `count` (if > 0) is honored
/// as a soft cap on transmitted probes.
pub struct RealPingOptions {
    pub dest: String,
    /// Number of probes to run. `Some(0)` or `None` => infinite (governed by puller).
    pub count: Option<usize>,
    /// Ignored here (cadence is pull-driven).
    pub interval_ms: Option<u64>,
    /// Ignored here (timeouts externalized to flow:stop_when).
    pub timeout_ms: u64,
    /// Starting sequence number (u16). Each probe increments it.
    pub seq_start: u16,
    /// Verbose diagnostics to stderr.
    pub verbose: bool,
}

pub fn spawn_iterator(opts: RealPingOptions) -> IteratorHandle {
    let _ = opts.interval_ms;
    let _ = opts.timeout_ms;

    #[cfg(unix)]
    {
        match crate::util::resolve_host(&opts.dest) {
            Ok(IpAddr::V4(ipv4)) => {
                match unix::PingSock::open(IpAddr::V4(ipv4), true, opts.verbose)
                    .or_else(|_| unix::PingSock::open(IpAddr::V4(ipv4), false, opts.verbose))
                {
                    Ok(sock) => {
                        let remaining = match opts.count {
                            Some(0) | None => None,
                            Some(n) => Some(n),
                        };
                        let iter = unix::PingIter::new(
                            sock,
                            opts.dest.clone(),
                            IpAddr::V4(ipv4),
                            opts.seq_start,
                            remaining,
                        );
                        return IteratorHandle {
                            kind: IteratorKind::Plugin(Arc::new(Mutex::new(iter))),
                        };
                    }
                    Err(e) => {
                        let row = error_row(&opts.dest, "0.0.0.0", 0, &format!("open socket: {}", e));
                        return IteratorHandle {
                            kind: IteratorKind::Plugin(Arc::new(Mutex::new(
                                ErrorOnceIter { once: Some(row) }
                            ))),
                        };
                    }
                }
            }
            Ok(IpAddr::V6(_)) => {
                let row = error_row(&opts.dest, "0.0.0.0", 0, "ICMPv6 is not implemented in this plugin");
                return IteratorHandle {
                    kind: IteratorKind::Plugin(Arc::new(Mutex::new(
                        ErrorOnceIter { once: Some(row) }
                    ))),
                };
            }
            Err(e) => {
                let row = error_row(&opts.dest, "0.0.0.0", 0, &e);
                return IteratorHandle {
                    kind: IteratorKind::Plugin(Arc::new(Mutex::new(
                        ErrorOnceIter { once: Some(row) }
                    ))),
                };
            }
        }
    }

    #[cfg(not(unix))]
    {
        let row = error_row(&opts.dest, "0.0.0.0", 0, "raw ICMP is not supported on this platform");
        IteratorHandle {
            kind: IteratorKind::Plugin(Arc::new(Mutex::new(
                ErrorOnceIter { once: Some(row) }
            ))),
        }
    }
}

/// Adapter: convert an Iterator handle into a 0-arg **transform** Function that
/// yields the next ping row per call (or returns "AGAIN"/"NO_MORE_DATA").
/// This mirrors the shape returned by `net:ping(...)`.
pub fn iter_to_transform(handle: IteratorHandle) -> Box<FunctionValue> {
    use mumu::parser::types::FunctionValue::RustClosure;

    Box::new(RustClosure(
        "net:real_ping-transform".to_string(),
        Arc::new(Mutex::new(
            move |_interp: &mut mumu::parser::interpreter::Interpreter, _args: Vec<Value>| {
                match &handle.kind {
                    IteratorKind::Core(state_arc) => {
                        let mut guard = state_arc
                            .lock()
                            .map_err(|_| "IteratorState lock error".to_string())?;
                        if guard.done || guard.current >= guard.end {
                            guard.done = true;
                            Err("NO_MORE_DATA".to_string())
                        } else {
                            let v = Value::Int(guard.current);
                            guard.current += 1;
                            if guard.current >= guard.end {
                                guard.done = true;
                            }
                            Ok(v)
                        }
                    }
                    IteratorKind::Plugin(plugin_arc) => {
                        let mut plugin = plugin_arc
                            .lock()
                            .map_err(|_| "Plugin Iterator lock error".to_string())?;
                        plugin.next_value()
                    }
                }
            },
        )),
        0,
    ))
}

fn error_row(dest: &str, ip: &str, seq: u16, message: &str) -> Value {
    let mut map = IndexMap::new();
    map.insert("dest".into(), Value::SingleString(dest.to_string()));
    map.insert("ip".into(), Value::SingleString(ip.to_string()));
    map.insert("seq".into(), Value::Int(seq as i32));
    map.insert("ok".into(), Value::Bool(false));
    map.insert("message".into(), Value::SingleString(message.to_string()));
    Value::KeyedArray(map)
}

#[cfg(unix)]
fn ok_row(dest: &str, ip: &str, seq: u16, bytes: usize, ttl: Option<u8>, ms: f64) -> Value {
    let mut map = IndexMap::new();
    map.insert("dest".into(), Value::SingleString(dest.to_string()));
    map.insert("ip".into(), Value::SingleString(ip.to_string()));
    map.insert("seq".into(), Value::Int(seq as i32));
    map.insert("ok".into(), Value::Bool(true));
    map.insert("bytes".into(), Value::Int(bytes.min(i32::MAX as usize) as i32));
    if let Some(t) = ttl {
        map.insert("ttl".into(), Value::Int(t as i32));
    }
    map.insert("ms".into(), Value::Float(ms));
    Value::KeyedArray(map)
}

#[cfg(unix)]
mod unix {
    use super::*;
    use libc;
    use std::io;
    use std::mem::{size_of, zeroed};
    use std::net::IpAddr;
    use std::time::Instant;

    // ICMP types
    const ICMP_ECHO: u8 = 8;
    const ICMP_ECHOREPLY: u8 = 0;

    // Linux-only: ICMP filter socket option constant (IPPROTO_ICMP level)
    #[cfg(target_os = "linux")]
    const ICMP_FILTER_OPT: libc::c_int = 1;

    #[cfg(target_os = "linux")]
    #[repr(C)]
    struct IcmpFilter {
        data: u32,
    }

    #[derive(Debug, Clone, Copy, PartialEq, Eq)]
    enum Mode {
        Datagram, // AF_INET, SOCK_DGRAM, IPPROTO_ICMP (no IP header on recv)
        Raw,      // AF_INET, SOCK_RAW,   IPPROTO_ICMP (includes IPv4 header)
    }

    pub(super) struct PingSock {
        fd: i32,
        mode: Mode,
        verbose: bool,
        // Only for Datagram: kernel-chosen ICMP Echo Identifier (sin_port).
        assigned_ident: Option<u16>,
    }

    impl Drop for PingSock {
        fn drop(&mut self) {
            unsafe { let _ = libc::close(self.fd); }
        }
    }

    impl PingSock {
        pub fn open(ip: IpAddr, prefer_datagram: bool, verbose: bool) -> Result<Self, String> {
            let v4 = match ip {
                IpAddr::V4(v) => v,
                IpAddr::V6(_) => return Err("ICMPv6 is not implemented".into()),
            };

            // sockaddr_in expects the address bytes in network order in memory.
            let mut addr: libc::sockaddr_in = unsafe { zeroed() };
            addr.sin_family = libc::AF_INET as libc::sa_family_t;
            addr.sin_port = 0;
            addr.sin_addr = libc::in_addr {
                s_addr: u32::from_ne_bytes(v4.octets()),
            };

            // Try datagram, then raw — **track which mode actually succeeded**.
            let (fd, used_mode) = if prefer_datagram {
                match Self::open_inner(Mode::Datagram, &addr, verbose) {
                    Ok(fd) => (fd, Mode::Datagram),
                    Err(e1) => {
                        if verbose {
                            eprintln!("[net:real_ping] datagram open failed: {e1} — falling back to RAW");
                        }
                        (Self::open_inner(Mode::Raw, &addr, verbose)?, Mode::Raw)
                    }
                }
            } else {
                match Self::open_inner(Mode::Raw, &addr, verbose) {
                    Ok(fd) => (fd, Mode::Raw),
                    Err(e1) => {
                        if verbose {
                            eprintln!("[net:real_ping] raw open failed: {e1} — falling back to DATAGRAM");
                        }
                        (Self::open_inner(Mode::Datagram, &addr, verbose)?, Mode::Datagram)
                    }
                }
            };

            // For datagram sockets, the kernel uses a per-socket Echo Identifier.
            // Read it with getsockname() (sin_port).
            let mut assigned_ident = None;
            if matches!(used_mode, Mode::Datagram) {
                unsafe {
                    let mut name: libc::sockaddr_in = zeroed();
                    let mut len: libc::socklen_t = size_of::<libc::sockaddr_in>() as libc::socklen_t;
                    let r = libc::getsockname(
                        fd,
                        &mut name as *mut libc::sockaddr_in as *mut libc::sockaddr,
                        &mut len,
                    );
                    if r == 0 {
                        // sin_port is network order
                        let id = u16::from_be(name.sin_port);
                        assigned_ident = Some(id);
                        if verbose {
                            eprintln!("[net:real_ping] datagram assigned ident (kernel) = {}", id);
                        }
                    } else if verbose {
                        eprintln!(
                            "[net:real_ping] warning: getsockname() failed: {}",
                            io::Error::last_os_error()
                        );
                    }
                }
            }

            if verbose {
                eprintln!(
                    "[net:real_ping] open => mode={:?}, fd={}, dest={}",
                    used_mode, fd, v4
                );
            }

            Ok(Self {
                fd,
                mode: used_mode,
                verbose,
                assigned_ident,
            })
        }

        fn open_inner(mode: Mode, addr: &libc::sockaddr_in, verbose: bool) -> Result<i32, String> {
            unsafe {
                let ty = match mode {
                    Mode::Datagram => libc::SOCK_DGRAM,
                    Mode::Raw => libc::SOCK_RAW,
                };
                let fd = libc::socket(libc::AF_INET, ty | libc::SOCK_NONBLOCK, libc::IPPROTO_ICMP);
                if fd < 0 {
                    return Err(io::Error::last_os_error().to_string());
                }

                // Connect the socket to the destination so that incoming ECHO REPLY
                // packets are routed to this socket (important for ping sockets).
                let rc = libc::connect(
                    fd,
                    addr as *const libc::sockaddr_in as *const libc::sockaddr,
                    std::mem::size_of::<libc::sockaddr_in>() as libc::socklen_t,
                );
                if rc != 0 && verbose {
                    eprintln!("[net:real_ping] warning: connect() failed: {}", io::Error::last_os_error());
                }

                // Linux: ICMP filter — allow only ECHOREPLY (best-effort; may require caps)
                #[cfg(target_os = "linux")]
                {
                    let mut f = IcmpFilter {
                        // Drop everything except ECHOREPLY
                        data: !((1u32) << (ICMP_ECHOREPLY as u32)),
                    };
                    let r = libc::setsockopt(
                        fd,
                        libc::IPPROTO_ICMP,
                        ICMP_FILTER_OPT,
                        &mut f as *mut _ as *const libc::c_void,
                        size_of::<IcmpFilter>() as libc::socklen_t,
                    );
                    if r != 0 && verbose {
                        eprintln!(
                            "[net:real_ping] warning: ICMP_FILTER setsockopt failed: {}",
                            io::Error::last_os_error()
                        );
                    }
                }

                // Try to get TTL via ancillary data on datagram sockets (best-effort).
                if matches!(mode, Mode::Datagram) {
                    let one: libc::c_int = 1;
                    let r = libc::setsockopt(
                        fd,
                        libc::IPPROTO_IP,
                        libc::IP_RECVTTL,
                        &one as *const _ as *const libc::c_void,
                        size_of::<libc::c_int>() as libc::socklen_t,
                    );
                    if r != 0 && verbose {
                        eprintln!(
                            "[net:real_ping] warning: IP_RECVTTL setsockopt failed: {}",
                            io::Error::last_os_error()
                        );
                    }
                }

                // Reasonable buffers
                let rcv = 128 * 1024;
                let snd = 64 * 1024;
                let _ = libc::setsockopt(
                    fd,
                    libc::SOL_SOCKET,
                    libc::SO_RCVBUF,
                    &rcv as *const _ as *const libc::c_void,
                    size_of::<libc::c_int>() as libc::socklen_t,
                );
                let _ = libc::setsockopt(
                    fd,
                    libc::SOL_SOCKET,
                    libc::SO_SNDBUF,
                    &snd as *const _ as *const libc::c_void,
                    size_of::<libc::c_int>() as libc::socklen_t,
                );

                Ok(fd)
            }
        }

        /// Effective identifier to use for filtering replies.
        #[inline]
        fn expected_ident(&self, user_ident: u16) -> u16 {
            match self.mode {
                Mode::Datagram => self.assigned_ident.unwrap_or(user_ident),
                Mode::Raw => user_ident,
            }
        }

        /// Send one Echo request (non-blocking). We use `send()` since the socket is connected.
        fn send_probe(&mut self, ident: u16, seq: u16) -> Result<(), String> {
            // ICMP header (8 bytes) + payload (u64 stamp + padding)
            let mut pkt = [0u8; 8 + 16];
            pkt[0] = ICMP_ECHO;
            pkt[1] = 0;
            // checksum later
            pkt[4..6].copy_from_slice(&ident.to_be_bytes());
            pkt[6..8].copy_from_slice(&seq.to_be_bytes());
            // payload (stamp monotonic nanos for completeness)
            let nanos = monotonic_nanos();
            pkt[8..16].copy_from_slice(&nanos.to_be_bytes());
            let csum = icmp_checksum(&pkt);
            pkt[2] = (csum >> 8) as u8;
            pkt[3] = (csum & 0xFF) as u8;

            let sent = unsafe { libc::send(self.fd, pkt.as_ptr() as *const libc::c_void, pkt.len(), 0) };
            if sent < 0 {
                return Err(io::Error::last_os_error().to_string());
            }
            if self.verbose {
                eprintln!(
                    "[net:real_ping] send => mode={:?} ident={} seq={} bytes={}",
                    self.mode,
                    ident,
                    seq,
                    pkt.len()
                );
            }
            Ok(())
        }

        /// Drain the socket: push all available matching replies into `ready`.
        fn recv_drain(
            &mut self,
            expected_ident: u16,
            inflight: &mut InflightRing,
            ready: &mut Vec<Value>,
            row_maker: &RowMaker,
            buf_rx: &mut [u8; 2048],
        ) {
            loop {
                match self.mode {
                    Mode::Datagram => {
                        let mut iov = libc::iovec {
                            iov_base: buf_rx.as_mut_ptr() as *mut libc::c_void,
                            iov_len: buf_rx.len(),
                        };
                        let mut cmsg_space = [0u8; 128];
                        let mut msg: libc::msghdr = unsafe { zeroed() };
                        msg.msg_name = std::ptr::null_mut();
                        msg.msg_namelen = 0;
                        msg.msg_iov = &mut iov;
                        msg.msg_iovlen = 1;
                        msg.msg_control = cmsg_space.as_mut_ptr() as *mut libc::c_void;
                        msg.msg_controllen = cmsg_space.len();

                        let r = unsafe { libc::recvmsg(self.fd, &mut msg as *mut libc::msghdr, 0) };
                        if r < 0 {
                            let err = io::Error::last_os_error();
                            if let Some(code) = err.raw_os_error() {
                                if code == libc::EAGAIN || code == libc::EWOULDBLOCK {
                                    break;
                                }
                            }
                            if self.verbose {
                                eprintln!("[net:real_ping] recvmsg(datagram) error: {}", err);
                            }
                            break;
                        }
                        let n = r as usize;
                        if n < 8 {
                            if self.verbose {
                                eprintln!("[net:real_ping] recvmsg(datagram) short={} bytes", n);
                            }
                            continue;
                        }

                        // TTL from ancillary data (if present)
                        let mut ttl_out: Option<u8> = None;
                        unsafe {
                            let mut cmsg = libc::CMSG_FIRSTHDR(&msg);
                            while !cmsg.is_null() {
                                if (*cmsg).cmsg_level == libc::IPPROTO_IP && (*cmsg).cmsg_type == libc::IP_TTL {
                                    let data_ptr = libc::CMSG_DATA(cmsg) as *const libc::c_uchar;
                                    ttl_out = Some(*data_ptr as u8);
                                    break;
                                }
                                cmsg = libc::CMSG_NXTHDR(&msg, cmsg);
                            }
                        }

                        let icmp = &buf_rx[..n];
                        if self.verbose {
                            eprintln!(
                                "[net:real_ping] recv(datagram) {} bytes; icmp_type={} exp_ident={}",
                                n, icmp[0], expected_ident
                            );
                        }
                        if icmp[0] != ICMP_ECHOREPLY || icmp[1] != 0 {
                            continue;
                        }
                        let r_ident = u16::from_be_bytes([icmp[4], icmp[5]]);
                        let r_seq = u16::from_be_bytes([icmp[6], icmp[7]]);
                        if r_ident != expected_ident {
                            if self.verbose {
                                eprintln!(
                                    "[net:real_ping] ident mismatch (got {}, want {})",
                                    r_ident, expected_ident
                                );
                            }
                            continue;
                        }

                        if let Some(sent) = inflight.take(r_seq) {
                            let rtt_ms = sent.elapsed().as_secs_f64() * 1000.0;
                            if self.verbose {
                                eprintln!(
                                    "[net:real_ping] OK seq={} rtt={:.3}ms bytes={} ttl={:?}",
                                    r_seq, rtt_ms, n, ttl_out
                                );
                            }
                            ready.push(row_maker.ok_row(r_seq, n, ttl_out, rtt_ms));
                        }
                    }
                    Mode::Raw => {
                        let r = unsafe {
                            libc::recv(self.fd, buf_rx.as_mut_ptr() as *mut libc::c_void, buf_rx.len(), 0)
                        };
                        if r < 0 {
                            let err = io::Error::last_os_error();
                            if let Some(code) = err.raw_os_error() {
                                if code == libc::EAGAIN || code == libc::EWOULDBLOCK {
                                    break;
                                }
                            }
                            if self.verbose {
                                eprintln!("[net:real_ping] recv(raw) error: {}", err);
                            }
                            break;
                        }
                        let n = r as usize;
                        if n < 28 {
                            if self.verbose {
                                eprintln!("[net:real_ping] recv(raw) short={} bytes", n);
                            }
                            continue;
                        }
                        let ihl = ((buf_rx[0] & 0x0F) * 4) as usize;
                        if n < ihl + 8 {
                            continue;
                        }
                        let ttl = buf_rx[8];
                        let icmp = &buf_rx[ihl..];
                        if self.verbose {
                            eprintln!(
                                "[net:real_ping] recv(raw) {} bytes; ihl={} icmp_type={} exp_ident={}",
                                n, ihl, icmp[0], expected_ident
                            );
                        }
                        if icmp[0] != ICMP_ECHOREPLY || icmp[1] != 0 {
                            continue;
                        }
                        let r_ident = u16::from_be_bytes([icmp[4], icmp[5]]);
                        let r_seq = u16::from_be_bytes([icmp[6], icmp[7]]);
                        if r_ident != expected_ident {
                            if self.verbose {
                                eprintln!(
                                    "[net:real_ping] ident mismatch (got {}, want {})",
                                    r_ident, expected_ident
                                );
                            }
                            continue;
                        }
                        if let Some(sent) = inflight.take(r_seq) {
                            let rtt_ms = sent.elapsed().as_secs_f64() * 1000.0;
                            if self.verbose {
                                eprintln!(
                                    "[net:real_ping] OK seq={} rtt={:.3}ms bytes={} ttl={}",
                                    r_seq, rtt_ms, n - ihl, ttl
                                );
                            }
                            ready.push(row_maker.ok_row(r_seq, n - ihl, Some(ttl), rtt_ms));
                        }
                    }
                }
            }
        }
    }

    #[inline]
    fn monotonic_nanos() -> u64 {
        Instant::now().elapsed().as_nanos() as u64
    }

    fn icmp_checksum(data: &[u8]) -> u16 {
        let mut sum: u32 = 0;
        let mut chunks = data.chunks_exact(2);
        for c in &mut chunks {
            let w = u16::from_be_bytes([c[0], c[1]]) as u32;
            sum = sum.wrapping_add(w);
        }
        if let [last] = chunks.remainder() {
            let w = u16::from_be_bytes([*last, 0]) as u32;
            sum = sum.wrapping_add(w);
        }
        while (sum >> 16) != 0 {
            sum = (sum & 0xFFFF) + (sum >> 16);
        }
        !(sum as u16)
    }

    /// Small power-of-two ring for inflight send timestamps (per seq).
    struct InflightRing {
        slots: [Option<Instant>; Self::CAP],
    }
    impl InflightRing {
        const CAP: usize = 1024; // power-of-two
        const MASK: usize = Self::CAP - 1;

        fn new() -> Self {
            Self { slots: [None; Self::CAP] }
        }
        #[inline]
        fn put(&mut self, seq: u16, t: Instant) {
            self.slots[(seq as usize) & Self::MASK] = Some(t);
        }
        #[inline]
        fn take(&mut self, seq: u16) -> Option<Instant> {
            let idx = (seq as usize) & Self::MASK;
            self.slots[idx].take()
        }
    }

    /// Helper to build rows quickly without re-allocating strings all the time.
    struct RowMaker {
        dest: String,
        ip_s: String,
    }
    impl RowMaker {
        fn new(dest: String, ip: IpAddr) -> Self {
            Self { dest, ip_s: ip.to_string() }
        }
        #[inline]
        fn ok_row(&self, seq: u16, bytes: usize, ttl: Option<u8>, ms: f64) -> Value {
            super::ok_row(&self.dest, &self.ip_s, seq, bytes, ttl, ms)
        }
    }

    #[derive(Clone, Copy)]
    struct IdSeq {
        ident: u16,
        next_seq: u16,
    }
    impl IdSeq {
        fn new(start_seq: u16) -> Self {
            Self { ident: default_ident(), next_seq: start_seq }
        }
        #[inline]
        fn next(&mut self) -> (u16, u16) {
            let s = self.next_seq;
            self.next_seq = self.next_seq.wrapping_add(1);
            (self.ident, s)
        }
    }

    #[inline]
    fn default_ident() -> u16 {
        (unsafe { libc::getpid() } as u16) ^ 0xBEEF
    }

    /// Pull-driven iterator: one probe per pull.
    pub(super) struct PingIter {
        sock: PingSock,
        ids: IdSeq,
        inflight: InflightRing,
        row_maker: RowMaker,
        ready: Vec<Value>,
        buf_rx: [u8; 2048],
        primed: bool,
        remaining: Option<usize>, // Some(n) => stop after n sends; None => unbounded
    }

    impl PingIter {
        pub fn new(
            sock: PingSock,
            dest: String,
            ip: IpAddr,
            seq_start: u16,
            remaining: Option<usize>,
        ) -> Self {
            if sock.verbose {
                eprintln!(
                    "[net:real_ping] iterator init => dest={} ip={} seq_start={} remaining={:?} mode={:?}",
                    dest, ip, seq_start, remaining, sock.mode
                );
            }

            Self {
                sock,
                ids: IdSeq::new(seq_start),
                inflight: InflightRing::new(),
                row_maker: RowMaker::new(dest, ip),
                ready: Vec::with_capacity(8),
                buf_rx: [0u8; 2048],
                primed: false,
                remaining,
            }
        }

        #[inline]
        fn can_send_more(&self) -> bool {
            self.remaining.map(|n| n > 0).unwrap_or(true)
        }

        #[inline]
        fn note_sent(&mut self) {
            if let Some(r) = self.remaining.as_mut() {
                if *r > 0 {
                    *r -= 1;
                }
            }
        }

        /// Send the next probe, recording inflight time if allowed.
        fn send_next(&mut self) -> Result<(), String> {
            if !self.can_send_more() {
                return Ok(()); // soft cap reached; do not send
            }
            let (ident, seq) = self.ids.next();
            self.sock.send_probe(ident, seq)?;
            self.inflight.put(seq, Instant::now());
            self.note_sent();
            Ok(())
        }
    }

    impl PluginIterator for PingIter {
        fn next_value(&mut self) -> Result<Value, String> {
            // Emit queued row first and opportunistically schedule the next probe.
            if let Some(v) = self.ready.pop() {
                let _ = self.send_next();
                return Ok(v);
            }

            // Drain socket (non-blocking).
            let exp_ident = self.sock.expected_ident(self.ids.ident);
            self.sock.recv_drain(
                exp_ident,
                &mut self.inflight,
                &mut self.ready,
                &self.row_maker,
                &mut self.buf_rx,
            );

            if let Some(v) = self.ready.pop() {
                let _ = self.send_next();
                return Ok(v);
            }

            // First call prime (or loss): send and ask to try again.
            if !self.primed {
                if self.can_send_more() {
                    if self.sock.verbose {
                        eprintln!("[net:real_ping] priming: first send");
                    }
                    self.send_next()?;
                    self.primed = true;
                    return Err("AGAIN".into());
                } else {
                    return Err("NO_MORE_DATA".into());
                }
            }

            if self.can_send_more() {
                let _ = self.send_next();
            }
            Err("AGAIN".into())
        }
    }
}

/// One error row on first `next_value()`, then EOF.
struct ErrorOnceIter {
    once: Option<Value>,
}

impl PluginIterator for ErrorOnceIter {
    fn next_value(&mut self) -> Result<Value, String> {
        match self.once.take() {
            Some(v) => Ok(v),
            None => Err("NO_MORE_DATA".into()),
        }
    }
}