supermachine 0.7.83

//! C1 — the owned loopback network stack for the native-execution VM (Stage 1;
//! see the Stage-1 spec doc).
//!
//! ## Why owned byte channels, not real TCP (the key design call)
//!
//! Warm-restore must reproduce a *live, established* guest↔guest connection in a
//! fresh process. The adversarial review's #1 risk was that serializing a
//! connected TCP control block (seq/snd_una/windows/wscale/timers — or smoltcp's
//! private equivalents) is the hardest part of Stage 1; gVisor, which runs real
//! TCP in its owned netstack, *still* cannot checkpoint a connected loopback
//! socket after 8 years.
//!
//! We sidestep it entirely: a **loopback** destination (127.0.0.0/8, ::1, an
//! intra-pool address) never traverses a wire — it is lossless, in-order, and the
//! guest **cannot observe TCP sequence numbers** (kernel-internal). So a loopback
//! `SOCK_STREAM` socket is backed by a supervisor-**owned reliable byte channel**,
//! not real TCP. This is semantically faithful for everything the guest observes
//! (stream bytes, connect/accept/listen, half-close + EOF, `getsockname`/
//! `getpeername`, flow-control backpressure) and is **trivially serializable**: a
//! connection's entire state is its 4-tuple + lifecycle state + the bytes buffered
//! in each direction. No TCB, no `smoltcp`, no private internals.
//!
//! Real-TCP **egress** (DNS, external HTTP) is a *separate plane* that stays on
//! host sockets (and external peers are not restorable on any system) — this
//! module owns only the loopback plane.
//!
//! ## Shape
//!
//! A connection is a pair of [`Sock`]s cross-linked by `peer`. Each end owns its
//! **receive** buffer (`rx`: bytes the peer has sent, awaiting this end's `recv`).
//! `send` on A appends to B's `rx` (bounded — backpressure when full, modelling
//! TCP flow control at the byte level so a guest can't OOM the supervisor); `recv`
//! drains this end's `rx`; `shutdown(WR)` on A marks B's `peer_wr_shut` so B reads
//! EOF once its `rx` drains. This module is **pure** (no supervisor hot-path
//! edits — the live SOCKET/BIND/CONNECT/ACCEPT + readiness wiring is C1b) and is
//! unit-tested standalone, including the capture→restore round-trip.

use std::collections::{HashMap, VecDeque};
use std::io::{self, Read, Write};

use crate::snapshot_frame::cap_count;

/// A dense socket identifier — the inner value of the C0 `SocketHandle` an fd maps
/// to (`FdTarget::LoopSocket`). Stable across capture/restore so the fd table's
/// references stay valid.
pub type SockId = u32;

// errno values returned (as `Err(errno)`); C1b maps these to the guest's `-errno`.
pub const EAGAIN: i32 = 11;
pub const EBADF: i32 = 9;
pub const EINVAL: i32 = 22;
pub const EPIPE: i32 = 32;
pub const EISCONN: i32 = 106;
pub const ENOTCONN: i32 = 107;
pub const ECONNREFUSED: i32 = 111;
pub const EADDRINUSE: i32 = 98;

/// Per-direction byte-buffer cap (flow-control bound; a full peer `rx` makes this
/// end non-writable, so a guest cannot grow supervisor memory without bound).
const RX_CAP: usize = 256 * 1024;

/// `shutdown(2)` `how`.
pub const SHUT_RD: i32 = 0;
pub const SHUT_WR: i32 = 1;
pub const SHUT_RDWR: i32 = 2;

/// A loopback endpoint (v4 left-justified into the 16-byte field, matching the
/// `LoopSnapshot` wire shape). `Eq`/`Hash` so it keys the listener map.
#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]
pub struct Endpoint {
    pub v6: bool,
    pub ip: [u8; 16],
    pub port: u16,
}

impl Endpoint {
    pub fn v4(a: [u8; 4], port: u16) -> Endpoint {
        let mut ip = [0u8; 16];
        ip[..4].copy_from_slice(&a);
        Endpoint {
            v6: false,
            ip,
            port,
        }
    }
}

/// A socket's lifecycle state. Owned-stream loopback collapses TCP's handshake
/// into instantaneous transitions (no wire), so there is no SYN_SENT/SYN_RCVD —
/// a `connect` to a live listener is `Established` immediately.
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
pub enum SockState {
    Unbound,
    Bound,
    Listen,
    Established,
    Closed,
}

/// One owned socket end.
#[derive(Clone, Debug, PartialEq, Eq)]
struct Sock {
    state: SockState,
    local: Option<Endpoint>,
    remote: Option<Endpoint>,
    /// The connected peer's id (an `Established` end always has one until the peer
    /// is dropped).
    peer: Option<SockId>,
    /// Bytes the peer sent, awaiting this end's `recv` (this end OWNS its rx).
    rx: VecDeque<u8>,
    /// This end called `shutdown(RD)` — further `recv` returns EOF, incoming data
    /// is discarded.
    rd_shut: bool,
    /// The peer closed its write side (or fully closed): once `rx` drains, this
    /// end's `recv` returns 0 (EOF) rather than `EAGAIN`.
    peer_wr_shut: bool,
    /// For a `Listen` socket: ids of the server-side ends of completed connections
    /// awaiting `accept`.
    backlog: VecDeque<SockId>,
    /// #36 cross-fork refcount: how many fd-table entries (across processes) name
    /// this socket. A fork-from-warm instance INHERITS the warm cell's listener (the
    /// fd-table snapshot copies the `FdVal::Loop`), so the listening socket is shared.
    /// `close` only tears the socket down (and deregisters its listener endpoint) when
    /// the LAST reference closes — so releasing one instance can't yank a listener the
    /// warm zygote (and the next acquire) still hold. Created at 1; the fork-table copy
    /// bumps it (`dup`), `close` drops it.
    refs: u32,
}

impl Sock {
    fn new() -> Sock {
        Sock {
            state: SockState::Unbound,
            local: None,
            remote: None,
            peer: None,
            rx: VecDeque::new(),
            rd_shut: false,
            peer_wr_shut: false,
            backlog: VecDeque::new(),
            refs: 1,
        }
    }
}

/// The supervisor-owned loopback network: all loopback sockets + the listener
/// index. One per pool/sandbox (the live wiring holds it behind a `Mutex`).
#[derive(Debug, Default)]
pub struct LoopNet {
    socks: HashMap<SockId, Sock>,
    listeners: HashMap<Endpoint, SockId>,
    next_id: SockId,
    next_ephemeral: u16,
}

impl LoopNet {
    pub fn new() -> LoopNet {
        LoopNet {
            socks: HashMap::new(),
            listeners: HashMap::new(),
            next_id: 1,
            next_ephemeral: 49152,
        }
    }

    fn alloc_id(&mut self) -> SockId {
        let id = self.next_id;
        self.next_id = self.next_id.checked_add(1).expect("SockId space exhausted");
        id
    }

    fn alloc_ephemeral(&mut self) -> u16 {
        // Ephemeral range 49152..=65535, wrapping. Loopback collisions are
        // tolerable (the 4-tuple disambiguates) for Stage 1.
        let p = self.next_ephemeral;
        self.next_ephemeral = if self.next_ephemeral >= 65535 {
            49152
        } else {
            self.next_ephemeral + 1
        };
        p
    }

    /// Create an unbound loopback stream socket.
    pub fn socket(&mut self) -> SockId {
        let id = self.alloc_id();
        self.socks.insert(id, Sock::new());
        id
    }

    fn get(&self, id: SockId) -> Result<&Sock, i32> {
        self.socks.get(&id).ok_or(EBADF)
    }
    fn get_mut(&mut self, id: SockId) -> Result<&mut Sock, i32> {
        self.socks.get_mut(&id).ok_or(EBADF)
    }

    pub fn bind(&mut self, id: SockId, mut ep: Endpoint) -> Result<(), i32> {
        // Wildcard bind (port 0): allocate a FREE ephemeral port — node's
        // `listen({port:0})`, libpq, and many servers bind port 0 and rely on
        // getsockname to learn the real port. Without this, every port-0 bind aliased
        // {ip, 0} so the 2nd collided (EADDRINUSE) and getsockname reported 0 (the
        // shopify-test mock servers all bind 127.0.0.1:0 → "EADDRINUSE 127.0.0.1").
        if ep.port == 0 {
            let mut chosen = 0u16;
            for _ in 0..16384 {
                let p = self.alloc_ephemeral();
                let cand = Endpoint { port: p, ..ep };
                if !self.listeners.contains_key(&cand) {
                    chosen = p;
                    break;
                }
            }
            if chosen == 0 {
                return Err(EADDRINUSE); // ephemeral range exhausted
            }
            ep.port = chosen;
        }
        if self.listeners.contains_key(&ep) {
            return Err(EADDRINUSE);
        }
        let s = self.get_mut(id)?;
        if s.state != SockState::Unbound {
            return Err(EINVAL);
        }
        s.local = Some(ep);
        s.state = SockState::Bound;
        Ok(())
    }

    pub fn listen(&mut self, id: SockId) -> Result<(), i32> {
        let ep = {
            let s = self.get_mut(id)?;
            match s.state {
                SockState::Bound | SockState::Listen => {}
                _ => return Err(EINVAL),
            }
            s.state = SockState::Listen;
            s.local.ok_or(EINVAL)?
        };
        self.listeners.insert(ep, id);
        Ok(())
    }

    /// The connected peer of an established socket (used by the integration layer
    /// to signal the peer's readiness eventfd after a `send`/`shutdown`/`close`).
    pub fn peer_of(&self, id: SockId) -> Option<SockId> {
        self.socks.get(&id).and_then(|s| s.peer)
    }

    /// Connect to a loopback listener. Instantaneous: creates the server-side end,
    /// cross-links the pair to `Established`, and enqueues the server end on the
    /// listener's backlog for `accept`. Returns the **listener's** id so the
    /// integration layer can wake a thread blocked in `accept` on it.
    pub fn connect(&mut self, id: SockId, dst: Endpoint) -> Result<SockId, i32> {
        {
            let s = self.get(id)?;
            if s.state == SockState::Established {
                return Err(EISCONN);
            }
            if !matches!(s.state, SockState::Unbound | SockState::Bound) {
                return Err(EINVAL);
            }
        }
        // Match an exact listener, else a WILDCARD (0.0.0.0 / ::) listener on the
        // same port+family: a server that binds INADDR_ANY (nginx's `0.0.0.0:80`)
        // must be reachable via a guest-local loopback connect (`127.0.0.1:80`),
        // exactly as on a real kernel where INADDR_ANY covers loopback. Linux also
        // treats an IPv6 wildcard listener as dual-stack by default
        // (IPV6_V6ONLY=0), so let IPv4 loopback connects find a same-port ::
        // listener too; BusyBox nc commonly binds this shape.
        let listener_id = self
            .listeners
            .get(&dst)
            .copied()
            .or_else(|| {
                let wild = Endpoint {
                    v6: dst.v6,
                    ip: [0u8; 16],
                    port: dst.port,
                };
                self.listeners.get(&wild).copied()
            })
            .or_else(|| {
                if dst.v6 {
                    None
                } else {
                    let wild_v6 = Endpoint {
                        v6: true,
                        ip: [0u8; 16],
                        port: dst.port,
                    };
                    self.listeners.get(&wild_v6).copied()
                }
            })
            .ok_or(ECONNREFUSED)?;
        // Local endpoint: keep an explicit bind, else allocate an ephemeral port on
        // the loopback address of the destination's family.
        let local = match self.get(id)?.local {
            Some(ep) => ep,
            None => {
                let port = self.alloc_ephemeral();
                if dst.v6 {
                    let mut ip = [0u8; 16];
                    ip[15] = 1; // ::1
                    Endpoint { v6: true, ip, port }
                } else {
                    Endpoint::v4([127, 0, 0, 1], port)
                }
            }
        };
        let server_id = self.alloc_id();
        // Server end.
        let mut server = Sock::new();
        server.state = SockState::Established;
        server.local = Some(dst);
        server.remote = Some(local);
        server.peer = Some(id);
        self.socks.insert(server_id, server);
        // Client end.
        {
            let c = self.get_mut(id)?;
            c.state = SockState::Established;
            c.local = Some(local);
            c.remote = Some(dst);
            c.peer = Some(server_id);
        }
        // Enqueue on the listener's backlog.
        self.get_mut(listener_id)?.backlog.push_back(server_id);
        Ok(listener_id)
    }

    /// Connect to the first registered listener. This preserves the legacy
    /// port-zero routing used by `Vm::connect()` and `expose_tcp(_, 0)`.
    pub fn connect_first(&mut self, id: SockId) -> Result<SockId, i32> {
        let dst = self.listeners.keys().copied().next().ok_or(ECONNREFUSED)?;
        self.connect(id, dst)
    }

    /// Pop a completed connection off a listener's backlog (the new server-side
    /// socket id). `EAGAIN` when none pending.
    pub fn accept(&mut self, listener_id: SockId) -> Result<SockId, i32> {
        let s = self.get_mut(listener_id)?;
        if s.state != SockState::Listen {
            return Err(EINVAL);
        }
        s.backlog.pop_front().ok_or(EAGAIN)
    }

    /// Append `buf` to the peer's rx (bounded). Returns bytes accepted (a short
    /// write under backpressure); `EAGAIN` if the peer rx is already full; `EPIPE`
    /// if the peer is gone or has shut its read side.
    pub fn send(&mut self, id: SockId, buf: &[u8]) -> Result<usize, i32> {
        let peer = {
            let s = self.get(id)?;
            if s.state != SockState::Established {
                return Err(ENOTCONN);
            }
            s.peer
        };
        let peer = peer.ok_or(EPIPE)?;
        let p = match self.socks.get_mut(&peer) {
            Some(p) if !p.rd_shut && p.state == SockState::Established => p,
            _ => return Err(EPIPE),
        };
        let space = RX_CAP.saturating_sub(p.rx.len());
        if space == 0 {
            return Err(EAGAIN);
        }
        let n = buf.len().min(space);
        p.rx.extend(&buf[..n]);
        Ok(n)
    }

    /// Drain up to `max` bytes from this end's rx. Empty + peer-write-shut ⇒ `Ok([])`
    /// (EOF); empty + peer still open ⇒ `EAGAIN`.
    pub fn recv(&mut self, id: SockId, max: usize) -> Result<Vec<u8>, i32> {
        let s = self.get_mut(id)?;
        if s.state != SockState::Established && s.state != SockState::Closed {
            return Err(ENOTCONN);
        }
        if s.rx.is_empty() {
            if s.peer_wr_shut || s.peer.is_none() {
                return Ok(Vec::new()); // EOF
            }
            return Err(EAGAIN);
        }
        let n = max.min(s.rx.len());
        Ok(s.rx.drain(..n).collect())
    }

    /// `shutdown(2)`. `SHUT_WR` tells the peer no more data is coming (peer reads
    /// EOF after draining); `SHUT_RD` stops this end reading.
    pub fn shutdown(&mut self, id: SockId, how: i32) -> Result<(), i32> {
        let peer = {
            let s = self.get_mut(id)?;
            if how == SHUT_RD || how == SHUT_RDWR {
                s.rd_shut = true;
                s.rx.clear();
            }
            s.peer
        };
        if (how == SHUT_WR || how == SHUT_RDWR) && peer.is_some() {
            if let Some(p) = self.socks.get_mut(&peer.unwrap()) {
                p.peer_wr_shut = true;
            }
        }
        Ok(())
    }

    /// Bump a socket's cross-fork refcount (the fd-table snapshot duplicated its
    /// `FdVal::Loop` into a forked child — see #36 / `Sock::refs`).
    pub fn dup(&mut self, id: SockId) {
        if let Some(s) = self.socks.get_mut(&id) {
            s.refs += 1;
        }
    }

    /// Close one reference to a socket. Returns `Ok(true)` when this was the LAST
    /// reference and the socket was actually torn down (peer detached, listener
    /// endpoint deregistered, socket removed), `Ok(false)` when other references (a
    /// fork sibling) still hold it and only the refcount was decremented. The caller
    /// (`loop_close`) frees the readiness eventfd + wakes the peer ONLY on a real
    /// teardown, so releasing one fork-from-warm instance can't strand a listener the
    /// warm zygote still owns.
    pub fn close(&mut self, id: SockId) -> Result<bool, i32> {
        {
            let s = self.socks.get_mut(&id).ok_or(EBADF)?;
            if s.refs > 1 {
                s.refs -= 1;
                return Ok(false); // still referenced elsewhere — keep it alive
            }
        }
        let s = self.socks.get(&id).ok_or(EBADF)?;
        let peer = s.peer;
        let local = s.local;
        let is_listener = s.state == SockState::Listen;
        if let Some(pid) = peer {
            if let Some(p) = self.socks.get_mut(&pid) {
                p.peer_wr_shut = true; // peer reads EOF
                p.peer = None; // further peer sends EPIPE
            }
        }
        if is_listener {
            if let Some(ep) = local {
                self.listeners.remove(&ep);
            }
        }
        self.socks.remove(&id);
        Ok(true)
    }

    /// Readiness predicate the C1b epoll/poll/select path consults for a loopback
    /// fd. Readable = data buffered, OR EOF pending (peer write-shut/gone), OR a
    /// listener with a pending connection.
    pub fn readable(&self, id: SockId) -> bool {
        match self.socks.get(&id) {
            Some(s) if s.state == SockState::Listen => !s.backlog.is_empty(),
            Some(s) => {
                !s.rx.is_empty()
                    || s.peer_wr_shut
                    || (s.state == SockState::Established && s.peer.is_none())
            }
            None => false,
        }
    }

    /// Writable = an established connection whose peer can still accept bytes
    /// (peer alive, read side open, rx not full).
    pub fn writable(&self, id: SockId) -> bool {
        let s = match self.socks.get(&id) {
            Some(s) if s.state == SockState::Established => s,
            _ => return false,
        };
        match s.peer.and_then(|p| self.socks.get(&p)) {
            Some(p) => !p.rd_shut && p.rx.len() < RX_CAP,
            None => false,
        }
    }

    pub fn local_addr(&self, id: SockId) -> Option<Endpoint> {
        self.socks.get(&id).and_then(|s| s.local)
    }
    pub fn peer_addr(&self, id: SockId) -> Option<Endpoint> {
        self.socks.get(&id).and_then(|s| s.remote)
    }

    // ─── capture / restore (the warm-restore round-trip) ─────────────────────

    /// Serialize the entire owned loopback world. Because both ends of every
    /// connection live here, this is complete and self-contained — no external
    /// peer to re-dial, no TCB to reconstruct.
    pub fn capture(&self) -> LoopSnapshot {
        let mut socks: Vec<SockRecord> = self
            .socks
            .iter()
            .map(|(&id, s)| SockRecord {
                id,
                state: s.state,
                local: s.local,
                remote: s.remote,
                peer: s.peer,
                rx: s.rx.iter().copied().collect(),
                rd_shut: s.rd_shut,
                peer_wr_shut: s.peer_wr_shut,
                backlog: s.backlog.iter().copied().collect(),
            })
            .collect();
        socks.sort_by_key(|r| r.id); // deterministic order for a stable snapshot
        LoopSnapshot {
            next_id: self.next_id,
            next_ephemeral: self.next_ephemeral,
            socks,
        }
    }

    /// Rebuild a `LoopNet` from a snapshot (the fresh-process restore).
    pub fn restore(snap: &LoopSnapshot) -> LoopNet {
        let mut net = LoopNet {
            socks: HashMap::new(),
            listeners: HashMap::new(),
            next_id: snap.next_id,
            next_ephemeral: snap.next_ephemeral,
        };
        for r in &snap.socks {
            let s = Sock {
                state: r.state,
                local: r.local,
                remote: r.remote,
                peer: r.peer,
                rx: r.rx.iter().copied().collect(),
                rd_shut: r.rd_shut,
                peer_wr_shut: r.peer_wr_shut,
                backlog: r.backlog.iter().copied().collect(),
                refs: 1, // fresh-process restore: one referencer; forks re-bump via dup
            };
            if s.state == SockState::Listen {
                if let Some(ep) = s.local {
                    net.listeners.insert(ep, r.id);
                }
            }
            net.socks.insert(r.id, s);
        }
        net
    }
}

/// The serializable form of one socket inside the on-disk [`LoopSnapshot`].
/// Public so C4 can bridge it; the field set is the *complete* state of an owned
/// loopback socket.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct SockRecord {
    pub id: SockId,
    pub state: SockState,
    pub local: Option<Endpoint>,
    pub remote: Option<Endpoint>,
    pub peer: Option<SockId>,
    pub rx: Vec<u8>,
    pub rd_shut: bool,
    pub peer_wr_shut: bool,
    pub backlog: Vec<SockId>,
}

/// A complete owned-loopback-world snapshot.
#[derive(Clone, Debug, PartialEq, Eq, Default)]
pub struct LoopSnapshot {
    pub next_id: SockId,
    pub next_ephemeral: u16,
    pub socks: Vec<SockRecord>,
}

// ─── LoopSnapshot codec (C4: the owned loopback state, serialized to bytes) ───
//
// Self-contained little-endian codec (no serde), mirroring state_snap's discipline:
// magic + version HARD-refusal, every count/blob capped before allocation
// (cap_count / LOOP_BLOB_MAX). C4 writes these bytes as the netstack section of
// state.snap; restore reads them back and rebuilds the LoopNet.

const LOOP_MAGIC: [u8; 8] = *b"SMLOOP01";
const LOOP_VERSION: u32 = 1;
const LOOP_BLOB_MAX: usize = 64 << 20;

#[inline]
fn rb<const N: usize, R: Read>(r: &mut R) -> io::Result<[u8; N]> {
    let mut x = [0u8; N];
    r.read_exact(&mut x)?;
    Ok(x)
}
#[inline]
fn r_u16<R: Read>(r: &mut R) -> io::Result<u16> {
    Ok(u16::from_le_bytes(rb::<2, _>(r)?))
}
#[inline]
fn r_u32<R: Read>(r: &mut R) -> io::Result<u32> {
    Ok(u32::from_le_bytes(rb::<4, _>(r)?))
}
#[inline]
fn ierr(m: &str) -> io::Error {
    io::Error::new(io::ErrorKind::InvalidData, m.to_string())
}
fn w_ep<W: Write>(w: &mut W, ep: &Option<Endpoint>) -> io::Result<()> {
    match ep {
        Some(e) => {
            w.write_all(&[1u8, e.v6 as u8])?;
            w.write_all(&e.ip)?;
            w.write_all(&e.port.to_le_bytes())
        }
        None => w.write_all(&[0u8]),
    }
}
fn r_ep<R: Read>(r: &mut R) -> io::Result<Option<Endpoint>> {
    if rb::<1, _>(r)?[0] == 0 {
        return Ok(None);
    }
    let v6 = rb::<1, _>(r)?[0] != 0;
    let ip = rb::<16, _>(r)?;
    let port = r_u16(r)?;
    Ok(Some(Endpoint { v6, ip, port }))
}

impl SockState {
    fn tag(&self) -> u8 {
        match self {
            SockState::Unbound => 0,
            SockState::Bound => 1,
            SockState::Listen => 2,
            SockState::Established => 3,
            SockState::Closed => 4,
        }
    }
    fn from_tag(t: u8) -> io::Result<SockState> {
        Ok(match t {
            0 => SockState::Unbound,
            1 => SockState::Bound,
            2 => SockState::Listen,
            3 => SockState::Established,
            4 => SockState::Closed,
            _ => return Err(ierr("bad SockState tag")),
        })
    }
}

impl SockRecord {
    fn write_to<W: Write>(&self, w: &mut W) -> io::Result<()> {
        w.write_all(&self.id.to_le_bytes())?;
        w.write_all(&[self.state.tag()])?;
        w_ep(w, &self.local)?;
        w_ep(w, &self.remote)?;
        match self.peer {
            Some(p) => {
                w.write_all(&[1u8])?;
                w.write_all(&p.to_le_bytes())?;
            }
            None => w.write_all(&[0u8])?,
        }
        w.write_all(&(self.rx.len() as u32).to_le_bytes())?;
        w.write_all(&self.rx)?;
        w.write_all(&[self.rd_shut as u8, self.peer_wr_shut as u8])?;
        w.write_all(&(self.backlog.len() as u32).to_le_bytes())?;
        for &b in &self.backlog {
            w.write_all(&b.to_le_bytes())?;
        }
        Ok(())
    }
    fn read_from<R: Read>(r: &mut R) -> io::Result<SockRecord> {
        let id = r_u32(r)?;
        let state = SockState::from_tag(rb::<1, _>(r)?[0])?;
        let local = r_ep(r)?;
        let remote = r_ep(r)?;
        let peer = if rb::<1, _>(r)?[0] != 0 {
            Some(r_u32(r)?)
        } else {
            None
        };
        let rxn = r_u32(r)? as usize;
        if rxn > LOOP_BLOB_MAX {
            return Err(ierr("loop rx blob too large"));
        }
        let mut rx = vec![0u8; rxn];
        r.read_exact(&mut rx)?;
        let [rd_shut, peer_wr_shut] = rb::<2, _>(r)?;
        let n = r_u32(r)? as usize;
        let cap = cap_count(n, 4, LOOP_BLOB_MAX);
        let mut backlog = Vec::with_capacity(cap);
        for _ in 0..n {
            backlog.push(r_u32(r)?);
        }
        Ok(SockRecord {
            id,
            state,
            local,
            remote,
            peer,
            rx,
            rd_shut: rd_shut != 0,
            peer_wr_shut: peer_wr_shut != 0,
            backlog,
        })
    }
}

impl LoopSnapshot {
    /// Serialize the owned loopback world to bytes (the netstack section of
    /// state.snap). Pairs with [`read_from`](Self::read_from), which HARD-refuses a
    /// bad magic / version (never a best-effort parse).
    pub fn write_to<W: Write>(&self, w: &mut W) -> io::Result<()> {
        w.write_all(&LOOP_MAGIC)?;
        w.write_all(&LOOP_VERSION.to_le_bytes())?;
        w.write_all(&self.next_id.to_le_bytes())?;
        w.write_all(&self.next_ephemeral.to_le_bytes())?;
        w.write_all(&(self.socks.len() as u32).to_le_bytes())?;
        for s in &self.socks {
            s.write_to(w)?;
        }
        Ok(())
    }
    pub fn read_from<R: Read>(r: &mut R) -> io::Result<LoopSnapshot> {
        if rb::<8, _>(r)? != LOOP_MAGIC {
            return Err(ierr("loop snapshot magic mismatch"));
        }
        if r_u32(r)? != LOOP_VERSION {
            return Err(ierr("loop snapshot version skew (hard refusal)"));
        }
        let next_id = r_u32(r)?;
        let next_ephemeral = r_u16(r)?;
        let n = r_u32(r)? as usize;
        // Min per-sock record: id(4)+state(1)+local(1)+remote(1)+peer(1)+rx_len(4)
        // +flags(2)+backlog_len(4) = 18 bytes.
        let cap = cap_count(n, 18, LOOP_BLOB_MAX);
        let mut socks = Vec::with_capacity(cap);
        for _ in 0..n {
            socks.push(SockRecord::read_from(r)?);
        }
        Ok(LoopSnapshot {
            next_id,
            next_ephemeral,
            socks,
        })
    }
}

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

    fn ep(p: u16) -> Endpoint {
        Endpoint::v4([127, 0, 0, 1], p)
    }

    /// Stand up a listener + a connected client/server pair; return (net, client, server).
    fn connected() -> (LoopNet, SockId, SockId) {
        let mut net = LoopNet::new();
        let lis = net.socket();
        net.bind(lis, ep(5432)).unwrap();
        net.listen(lis).unwrap();
        let cli = net.socket();
        net.connect(cli, ep(5432)).unwrap();
        let srv = net.accept(lis).unwrap();
        (net, cli, srv)
    }

    #[test]
    fn connect_accept_bidirectional_stream() {
        let (mut net, cli, srv) = connected();
        assert_eq!(net.send(cli, b"ping").unwrap(), 4);
        assert_eq!(net.recv(srv, 64).unwrap(), b"ping");
        assert_eq!(net.send(srv, b"pong!").unwrap(), 5);
        assert_eq!(net.recv(cli, 64).unwrap(), b"pong!");
        // 4-tuple cross-check.
        assert_eq!(net.peer_addr(cli).unwrap(), ep(5432));
        assert_eq!(net.local_addr(srv).unwrap(), ep(5432));
        assert_eq!(net.peer_addr(srv).unwrap(), net.local_addr(cli).unwrap());
    }

    #[test]
    fn recv_empty_is_eagain_then_eof_after_peer_shutdown() {
        let (mut net, cli, srv) = connected();
        assert_eq!(net.recv(srv, 64), Err(EAGAIN)); // nothing yet
        net.send(cli, b"last").unwrap();
        net.shutdown(cli, SHUT_WR).unwrap();
        assert_eq!(net.recv(srv, 64).unwrap(), b"last"); // buffered data first
        assert_eq!(net.recv(srv, 64).unwrap(), b""); // THEN EOF (not EAGAIN)
    }

    #[test]
    fn connect_with_no_listener_is_refused() {
        let mut net = LoopNet::new();
        let cli = net.socket();
        assert_eq!(net.connect(cli, ep(9999)), Err(ECONNREFUSED));
    }

    #[test]
    fn connect_first_routes_to_registered_listener() {
        let mut net = LoopNet::new();
        let lis = net.socket();
        net.bind(lis, ep(8080)).unwrap();
        net.listen(lis).unwrap();
        let cli = net.socket();
        assert_eq!(net.connect_first(cli).unwrap(), lis);
        let srv = net.accept(lis).unwrap();
        net.send(cli, b"first").unwrap();
        assert_eq!(net.recv(srv, 16).unwrap(), b"first");
    }

    #[test]
    fn ipv4_loopback_reaches_ipv6_wildcard_listener() {
        let mut net = LoopNet::new();
        let lis = net.socket();
        net.bind(
            lis,
            Endpoint {
                v6: true,
                ip: [0u8; 16],
                port: 18080,
            },
        )
        .unwrap();
        net.listen(lis).unwrap();
        let cli = net.socket();
        assert_eq!(net.connect(cli, ep(18080)).unwrap(), lis);
        let srv = net.accept(lis).unwrap();
        net.send(cli, b"GET / HTTP/1.0\r\n\r\n").unwrap();
        assert_eq!(net.recv(srv, 64).unwrap(), b"GET / HTTP/1.0\r\n\r\n");
    }

    #[test]
    fn send_after_peer_close_is_epipe_and_peer_reads_eof() {
        let (mut net, cli, srv) = connected();
        net.send(cli, b"buffered").unwrap();
        net.close(srv).unwrap();
        // Peer (srv) gone: cli.send => EPIPE; cli.recv => EOF.
        assert_eq!(net.send(cli, b"x"), Err(EPIPE));
        assert_eq!(net.recv(cli, 64).unwrap(), b""); // EOF (peer_wr_shut set on close)
    }

    #[test]
    fn flow_control_backpressure_and_writable() {
        let (mut net, cli, srv) = connected();
        // Fill the peer's rx to the cap.
        let big = vec![0u8; RX_CAP];
        assert_eq!(net.send(cli, &big).unwrap(), RX_CAP);
        assert!(!net.writable(cli), "peer rx full ⇒ not writable");
        assert_eq!(net.send(cli, b"more"), Err(EAGAIN)); // full
                                                         // Drain on the server side re-opens the window.
        let _ = net.recv(srv, RX_CAP).unwrap();
        assert!(net.writable(cli), "drained ⇒ writable again");
    }

    #[test]
    fn readability_predicate() {
        let mut net = LoopNet::new();
        let lis = net.socket();
        net.bind(lis, ep(80)).unwrap();
        net.listen(lis).unwrap();
        assert!(!net.readable(lis), "no pending conn");
        let cli = net.socket();
        net.connect(cli, ep(80)).unwrap();
        assert!(net.readable(lis), "listener readable ⇒ accept won't block");
        let srv = net.accept(lis).unwrap();
        assert!(!net.readable(srv), "fresh conn, no data");
        net.send(cli, b"q").unwrap();
        assert!(net.readable(srv), "data buffered ⇒ readable");
    }

    /// The headline warm-restore test: a connection with bytes in flight in BOTH
    /// directions is captured, the net is dropped, and a FRESH LoopNet is restored
    /// — the SAME connection continues sequence-correctly (buffered bytes intact,
    /// both directions, EOF semantics preserved).
    #[test]
    fn capture_restore_round_trip_preserves_live_connection() {
        let (mut net, cli, srv) = connected();
        // In-flight bytes in both directions (unread at capture).
        net.send(cli, b"client->server in flight").unwrap();
        net.send(srv, b"server->client in flight").unwrap();
        // Half-close one direction to prove the flag survives.
        net.shutdown(srv, SHUT_WR).unwrap();

        let snap = net.capture();
        // Structural equality of the snapshot itself round-trips.
        let mut buf_round = LoopNet::restore(&snap).capture();
        buf_round.socks.sort_by_key(|r| r.id);
        assert_eq!(snap, buf_round, "snapshot is restore-stable");

        // Drop the live net; rebuild fresh.
        drop(net);
        let mut r = LoopNet::restore(&snap);

        // The client still reads the server's in-flight bytes, THEN EOF (the
        // half-close survived).
        assert_eq!(r.recv(cli, 64).unwrap(), b"server->client in flight");
        assert_eq!(
            r.recv(cli, 64).unwrap(),
            b"",
            "peer_wr_shut survived restore ⇒ EOF"
        );
        // The server still reads the client's in-flight bytes.
        assert_eq!(r.recv(srv, 64).unwrap(), b"client->server in flight");
        // And the connection keeps working: client can still send, server reads it.
        assert_eq!(r.send(cli, b"after restore").unwrap(), 13);
        assert_eq!(r.recv(srv, 64).unwrap(), b"after restore");
        // Endpoints (4-tuple) preserved.
        assert_eq!(r.peer_addr(cli).unwrap(), ep(5432));
    }

    #[test]
    fn listener_and_backlog_survive_restore() {
        let mut net = LoopNet::new();
        let lis = net.socket();
        net.bind(lis, ep(8080)).unwrap();
        net.listen(lis).unwrap();
        let cli = net.socket();
        net.connect(cli, ep(8080)).unwrap(); // pending in backlog, not yet accepted
        let snap = net.capture();
        let mut r = LoopNet::restore(&snap);
        // The pending connection is still acceptable after restore.
        let srv = r.accept(lis).expect("backlog survived");
        r.send(cli, b"hi").unwrap();
        assert_eq!(r.recv(srv, 64).unwrap(), b"hi");
        // The listener is re-registered (a NEW connect finds it).
        let cli2 = r.socket();
        assert!(
            r.connect(cli2, ep(8080)).is_ok(),
            "listener re-indexed after restore"
        );
    }

    /// C4 building block: the owned loopback world survives a BYTE round-trip
    /// (capture → write_to → read_from → restore) — the connection, in-flight bytes
    /// (both directions), half-close, and a pending listener backlog all intact.
    #[test]
    fn loop_snapshot_bytes_round_trip() {
        let (mut net, cli, srv) = connected();
        net.send(cli, b"in flight c->s").unwrap();
        net.send(srv, b"in flight s->c").unwrap();
        net.shutdown(srv, SHUT_WR).unwrap();
        // plus a listener with a pending (unaccepted) backlog.
        let lis = net.socket();
        net.bind(lis, ep(9000)).unwrap();
        net.listen(lis).unwrap();
        let c2 = net.socket();
        net.connect(c2, ep(9000)).unwrap();

        let snap = net.capture();
        let mut buf = Vec::new();
        snap.write_to(&mut buf).unwrap();
        let back = LoopSnapshot::read_from(&mut std::io::Cursor::new(&buf[..])).unwrap();
        assert_eq!(snap, back, "LoopSnapshot survives the byte round-trip");

        // Restore from the DESERIALIZED snapshot → the connection keeps working.
        let mut r = LoopNet::restore(&back);
        assert_eq!(r.recv(cli, 64).unwrap(), b"in flight s->c");
        assert_eq!(
            r.recv(cli, 64).unwrap(),
            b"",
            "half-close survived the byte round-trip"
        );
        assert_eq!(r.recv(srv, 64).unwrap(), b"in flight c->s");
        let srv2 = r.accept(lis).expect("backlog survived the byte round-trip");
        r.send(c2, b"hi2").unwrap();
        assert_eq!(r.recv(srv2, 64).unwrap(), b"hi2");
    }

    #[test]
    fn loop_snapshot_rejects_bad_magic_and_version() {
        let mut buf = Vec::new();
        LoopNet::new().capture().write_to(&mut buf).unwrap();
        let mut bad_ver = buf.clone();
        bad_ver[8] ^= 0xff; // corrupt the version (right after the 8-byte magic)
        assert!(LoopSnapshot::read_from(&mut std::io::Cursor::new(&bad_ver[..])).is_err());
        let mut bad_magic = buf.clone();
        bad_magic[0] ^= 0xff;
        assert!(LoopSnapshot::read_from(&mut std::io::Cursor::new(&bad_magic[..])).is_err());
    }
}