diff --git a/src/cluster.rs b/src/cluster.rs index b5953ad..950900c 100644 --- a/src/cluster.rs +++ b/src/cluster.rs @@ -1,8 +1,8 @@ //! RFC 010 — clustering (smarm⇄smarm, explicit remote boundary). //! -//! c1: feature flag + optional deps only. The owned envelope (c2), transport -//! trait (c3), and everything above them land in later chunks. This module is -//! intentionally empty so the `cluster` feature's default-build invariance is -//! reviewable in isolation. +//! c1: feature flag + optional deps. c2: the owned envelope. c3: the +//! transport trait (control connection), framed codec, and the TCP + +//! loopback impls. Everything above them lands in later chunks. pub mod envelope; +pub mod transport; diff --git a/src/cluster/transport.rs b/src/cluster/transport.rs new file mode 100644 index 0000000..f16057d --- /dev/null +++ b/src/cluster/transport.rs @@ -0,0 +1,194 @@ +//! RFC 010 c3 — transport abstraction for the **control** connection. +//! +//! Scope, per RFC 010 v2 §5 and D2: +//! +//! - A "connection" here is the *control* connection: the one carrying this +//! RFC's frame inventory ([`crate::cluster::envelope::Frame`]), whose +//! heartbeats feed failure detection. The trait deliberately says nothing +//! about how many connections a peer pair may hold — the jarred rkyv bulk +//! plane opens **additional per-peer connections** outside this trait, and +//! nothing here may foreclose that. +//! - Homogeneous smarm⇄smarm only. The BEAM membrane is *not* a transport +//! impl and the trait does not accommodate it (D2). +//! - Addresses are opaque, **pre-resolved** strings. Name resolution is a +//! single separate seam (roadmap c9); impls reject unresolved names rather +//! than resolving them. +//! +//! Blocking model: [`Conn`] calls block the caller. The TCP impl parks the +//! calling *actor* (fd readiness via the scheduler); the loopback impl blocks +//! the calling *OS thread* and is a test transport — do not drive it from a +//! scheduler thread. +//! +//! Framing is not part of the trait: [`FramedConn`] is the single shared +//! codec that turns any byte-stream [`Conn`] into a frame pipe, feeding +//! [`Frame::decode`]'s incremental contract. Impls never re-implement +//! framing, and the conformance suite exercises the same codec over every +//! impl. + +use std::io; + +use crate::cluster::envelope::{DecodeError, EncodeError, Frame}; + +pub mod loopback; +pub mod tcp; + +/// An established control connection: a bidirectional byte stream. +pub trait Conn: Send { + /// Read at least one byte, blocking the caller until data is available, + /// EOF, or error. `Ok(0)` means EOF: the peer closed and all bytes it + /// wrote before closing have been consumed. + fn read(&mut self, buf: &mut [u8]) -> io::Result; + + /// Write the whole buffer, blocking the caller as needed. + fn write_all(&mut self, buf: &[u8]) -> io::Result<()>; + + /// Close both directions. Idempotent. Bytes already written remain + /// readable at the peer, which then observes EOF; peer writes after this + /// fail. + fn close(&mut self); + + /// Diagnostic label for logs only. Mesh identity comes from the + /// handshake (`Hello`/`HelloAck`), never from the transport. + fn peer_addr(&self) -> String; +} + +/// A bound listen point producing inbound [`Conn`]s. +pub trait Listener: Send { + /// Accept the next inbound connection, blocking the caller. + fn accept(&mut self) -> io::Result>; + + /// The concrete bound address, dialable as-is (e.g. the real port when + /// bound with port 0). + fn local_addr(&self) -> String; +} + +/// A way of establishing control connections. Object-safe on purpose: the +/// connector and membership layers hold `&dyn Transport` / boxed conns +/// rather than growing a generic parameter. +pub trait Transport: Send + Sync { + /// Connect to a peer's listen address. Blocks the caller until + /// established or failed. + fn dial(&self, addr: &str) -> io::Result>; + + /// Bind a listen point. + fn listen(&self, addr: &str) -> io::Result>; +} + +impl std::fmt::Debug for dyn Conn { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Conn({})", self.peer_addr()) + } +} + +impl std::fmt::Debug for dyn Listener { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + write!(f, "Listener({})", self.local_addr()) + } +} + +/// Error surface of [`FramedConn::send`]. +#[derive(Debug)] +pub enum SendError { + /// The frame could not be encoded (e.g. a field over its wire limit). + Encode(EncodeError), + /// The transport failed mid-write. + Io(io::Error), +} + +impl std::fmt::Display for SendError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + SendError::Encode(e) => write!(f, "frame encode failed: {e:?}"), + SendError::Io(e) => write!(f, "transport write failed: {e}"), + } + } +} + +impl std::error::Error for SendError {} + +/// Error surface of [`FramedConn::recv`]. +#[derive(Debug)] +pub enum RecvError { + /// The byte stream is not a valid frame stream (bad tag, lying length, + /// oversized frame, …). The connection is unusable. + Corrupt(DecodeError), + /// The peer closed mid-frame: EOF arrived with a partial frame buffered. + /// Distinct from a clean close, which is `Ok(None)`. + TruncatedByPeer, + /// The transport failed mid-read. + Io(io::Error), +} + +impl std::fmt::Display for RecvError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + RecvError::Corrupt(e) => write!(f, "frame stream corrupt: {e:?}"), + RecvError::TruncatedByPeer => write!(f, "peer closed mid-frame"), + RecvError::Io(e) => write!(f, "transport read failed: {e}"), + } + } +} + +impl std::error::Error for RecvError {} + +/// How many bytes each blocking read asks the transport for. +const READ_CHUNK: usize = 8 * 1024; + +/// The shared framed codec: one of these per control connection, owning the +/// [`Conn`] and the reassembly buffer. Frames may arrive split or coalesced +/// arbitrarily; [`recv`](FramedConn::recv) reassembles either way. +pub struct FramedConn { + conn: Box, + rbuf: Vec, +} + +impl FramedConn { + pub fn new(conn: Box) -> Self { + FramedConn { + conn, + rbuf: Vec::new(), + } + } + + /// Encode and write one frame. + pub fn send(&mut self, frame: &Frame) -> Result<(), SendError> { + let mut out = Vec::new(); + frame.encode(&mut out).map_err(SendError::Encode)?; + self.conn.write_all(&out).map_err(SendError::Io) + } + + /// Receive the next frame. `Ok(None)` is a clean close: EOF at a frame + /// boundary. EOF mid-frame is [`RecvError::TruncatedByPeer`]. + pub fn recv(&mut self) -> Result, RecvError> { + loop { + match Frame::decode(&self.rbuf) { + Ok(Some((frame, consumed))) => { + self.rbuf.drain(..consumed); + return Ok(Some(frame)); + } + Ok(None) => {} + Err(e) => return Err(RecvError::Corrupt(e)), + } + let mut chunk = [0u8; READ_CHUNK]; + let n = self.conn.read(&mut chunk).map_err(RecvError::Io)?; + if n == 0 { + return if self.rbuf.is_empty() { + Ok(None) + } else { + Err(RecvError::TruncatedByPeer) + }; + } + self.rbuf.extend_from_slice(&chunk[..n]); + } + } + + /// Close the underlying connection (idempotent, see [`Conn::close`]). + pub fn close(&mut self) { + self.conn.close(); + } + + /// Diagnostic label of the underlying connection. + pub fn peer_addr(&self) -> String { + self.conn.peer_addr() + } +} diff --git a/src/cluster/transport/loopback.rs b/src/cluster/transport/loopback.rs new file mode 100644 index 0000000..61705f9 --- /dev/null +++ b/src/cluster/transport/loopback.rs @@ -0,0 +1,274 @@ +//! In-memory loopback transport — a shipped **test** transport. +//! +//! Lets Phases 2–4 exercise protocol logic (connector, membership, +//! monitors) through the real transport trait and the real framed codec +//! without sockets or timing flake. +//! +//! Blocking model: calls block the **OS thread** on a condvar. That is the +//! right shape for plain `#[test]`s driving protocol state machines; it is +//! the wrong shape for scheduler threads. Do not drive a loopback conn from +//! inside an actor — use the TCP impl there. +//! +//! Semantics mirror TCP shutdown where it matters for the codec: bytes +//! written before `close` remain readable at the peer, which then sees EOF; +//! writes toward a closed peer fail with `BrokenPipe`. Write buffers are +//! unbounded, so writes never block — backpressure is not simulated. + +use std::collections::{HashMap, VecDeque}; +use std::io; +use std::sync::{Arc, Condvar, Mutex, MutexGuard}; + +use super::{Conn, Listener, Transport}; + +/// Poison-tolerant lock: a panicked holder in a *test* transport must not +/// cascade; the byte-queue state stays consistent under every early return. +fn lock(m: &Mutex) -> MutexGuard<'_, T> { + match m.lock() { + Ok(g) => g, + Err(poisoned) => poisoned.into_inner(), + } +} + +// --------------------------------------------------------------------------- +// One direction of a duplex: a byte queue with close flags for both ends +// --------------------------------------------------------------------------- + +#[derive(Default)] +struct PipeState { + bytes: VecDeque, + /// The writing end closed: readers drain remaining bytes, then EOF. + write_closed: bool, + /// The reading end closed: writers fail with `BrokenPipe`. + read_closed: bool, +} + +#[derive(Default)] +struct Pipe { + state: Mutex, + cv: Condvar, +} + +impl Pipe { + fn write_all(&self, buf: &[u8]) -> io::Result<()> { + let mut st = lock(&self.state); + if st.write_closed { + return Err(io::Error::new( + io::ErrorKind::NotConnected, + "loopback conn closed locally", + )); + } + if st.read_closed { + return Err(io::Error::new( + io::ErrorKind::BrokenPipe, + "loopback peer closed", + )); + } + st.bytes.extend(buf); + self.cv.notify_all(); + Ok(()) + } + + fn read(&self, buf: &mut [u8]) -> io::Result { + if buf.is_empty() { + return Ok(0); + } + let mut st = lock(&self.state); + loop { + if !st.bytes.is_empty() { + let n = st.bytes.len().min(buf.len()); + for (slot, byte) in buf.iter_mut().zip(st.bytes.drain(..n)) { + *slot = byte; + } + return Ok(n); + } + if st.write_closed || st.read_closed { + return Ok(0); // EOF: peer closed, or our own end closed. + } + st = match self.cv.wait(st) { + Ok(g) => g, + Err(poisoned) => poisoned.into_inner(), + }; + } + } + + /// Close from the writer side: remaining bytes stay readable, then EOF. + fn close_write(&self) { + lock(&self.state).write_closed = true; + self.cv.notify_all(); + } + + /// Close from the reader side: peer writes fail from now on. + fn close_read(&self) { + lock(&self.state).read_closed = true; + self.cv.notify_all(); + } +} + +// --------------------------------------------------------------------------- +// Conn: two pipes, one per direction +// --------------------------------------------------------------------------- + +/// One end of an established loopback connection. +pub struct LoopbackConn { + tx: Arc, + rx: Arc, + peer: String, +} + +impl Conn for LoopbackConn { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + self.rx.read(buf) + } + + fn write_all(&mut self, buf: &[u8]) -> io::Result<()> { + self.tx.write_all(buf) + } + + fn close(&mut self) { + self.tx.close_write(); + self.rx.close_read(); + } + + fn peer_addr(&self) -> String { + self.peer.clone() + } +} + +impl Drop for LoopbackConn { + fn drop(&mut self) { + self.close(); + } +} + +fn conn_pair(listen_addr: &str, conn_no: u64) -> (LoopbackConn, LoopbackConn) { + let a_to_b = Arc::new(Pipe::default()); + let b_to_a = Arc::new(Pipe::default()); + let dialer = LoopbackConn { + tx: a_to_b.clone(), + rx: b_to_a.clone(), + peer: listen_addr.to_string(), + }; + let accepted = LoopbackConn { + tx: b_to_a, + rx: a_to_b, + peer: format!("{listen_addr}#dialer-{conn_no}"), + }; + (dialer, accepted) +} + +// --------------------------------------------------------------------------- +// Listener + registry +// --------------------------------------------------------------------------- + +#[derive(Default)] +struct AcceptState { + pending: VecDeque, + closed: bool, +} + +#[derive(Default)] +struct AcceptQueue { + state: Mutex, + cv: Condvar, +} + +/// A bound loopback listen point. +pub struct LoopbackListener { + addr: String, + queue: Arc, + registry: Arc>, +} + +impl Listener for LoopbackListener { + fn accept(&mut self) -> io::Result> { + let mut st = lock(&self.queue.state); + loop { + if let Some(conn) = st.pending.pop_front() { + return Ok(Box::new(conn)); + } + if st.closed { + return Err(io::Error::new( + io::ErrorKind::NotConnected, + "loopback listener closed", + )); + } + st = match self.queue.cv.wait(st) { + Ok(g) => g, + Err(poisoned) => poisoned.into_inner(), + }; + } + } + + fn local_addr(&self) -> String { + self.addr.clone() + } +} + +impl Drop for LoopbackListener { + fn drop(&mut self) { + lock(&self.registry).listeners.remove(&self.addr); + let mut st = lock(&self.queue.state); + st.closed = true; + self.queue.cv.notify_all(); + } +} + +#[derive(Default)] +struct Registry { + listeners: HashMap>, + dial_count: u64, +} + +/// The loopback transport. Addresses are arbitrary strings scoped to one +/// transport instance; distinct instances never see each other's listeners. +#[derive(Default)] +pub struct LoopbackTransport { + registry: Arc>, +} + +impl Transport for LoopbackTransport { + fn dial(&self, addr: &str) -> io::Result> { + let (queue, conn_no) = { + let mut reg = lock(&self.registry); + reg.dial_count += 1; + let no = reg.dial_count; + match reg.listeners.get(addr) { + Some(q) => (q.clone(), no), + None => { + return Err(io::Error::new( + io::ErrorKind::ConnectionRefused, + format!("no loopback listener at {addr:?}"), + )); + } + } + }; + let (dialer, accepted) = conn_pair(addr, conn_no); + let mut st = lock(&queue.state); + if st.closed { + return Err(io::Error::new( + io::ErrorKind::ConnectionRefused, + format!("loopback listener at {addr:?} closed"), + )); + } + st.pending.push_back(accepted); + queue.cv.notify_all(); + Ok(Box::new(dialer)) + } + + fn listen(&self, addr: &str) -> io::Result> { + let queue = Arc::new(AcceptQueue::default()); + let mut reg = lock(&self.registry); + if reg.listeners.contains_key(addr) { + return Err(io::Error::new( + io::ErrorKind::AddrInUse, + format!("loopback listener already bound at {addr:?}"), + )); + } + reg.listeners.insert(addr.to_string(), queue.clone()); + Ok(Box::new(LoopbackListener { + addr: addr.to_string(), + queue, + registry: self.registry.clone(), + })) + } +} diff --git a/src/cluster/transport/tcp.rs b/src/cluster/transport/tcp.rs new file mode 100644 index 0000000..c725088 --- /dev/null +++ b/src/cluster/transport/tcp.rs @@ -0,0 +1,277 @@ +//! TCP transport — the production control-plane transport. +//! +//! Blocking model: every blocking point parks the **calling actor** on fd +//! readiness ([`crate::scheduler::wait_readable`] / `wait_writable`); the +//! scheduler thread is never blocked. All conn/listener methods must +//! therefore run inside an actor. `listen` itself only binds (no waiting) +//! and is callable anywhere. +//! +//! Addresses are pre-resolved `ip:port` strings (`SocketAddr` syntax, IPv4 +//! or IPv6). Hostnames are rejected with `InvalidInput`: name resolution is +//! the single c9 seam, not something each transport does on the side. +//! +//! Writes use `send(2)` with `MSG_NOSIGNAL` — a peer reset must surface as +//! `BrokenPipe`/`ConnectionReset`, not `SIGPIPE`. + +use std::io; +use std::net::{SocketAddr, TcpListener as StdListener, TcpStream}; +use std::os::fd::{AsRawFd, RawFd}; + +use crate::scheduler::{wait_readable, wait_writable}; + +use super::{Conn, Listener, Transport}; + +// --------------------------------------------------------------------------- +// sockaddr plumbing +// --------------------------------------------------------------------------- + +/// A `sockaddr_in`/`sockaddr_in6` built from a parsed `SocketAddr`, plus its +/// length, ready for `connect(2)`. +union SockAddrUnion { + v4: libc::sockaddr_in, + v6: libc::sockaddr_in6, +} + +fn to_sockaddr(sa: &SocketAddr) -> (SockAddrUnion, libc::socklen_t) { + match sa { + SocketAddr::V4(v4) => { + let raw = libc::sockaddr_in { + sin_family: libc::AF_INET as libc::sa_family_t, + sin_port: v4.port().to_be(), + sin_addr: libc::in_addr { + s_addr: u32::from_be_bytes(v4.ip().octets()).to_be(), + }, + sin_zero: [0; 8], + }; + ( + SockAddrUnion { v4: raw }, + std::mem::size_of::() as libc::socklen_t, + ) + } + SocketAddr::V6(v6) => { + let raw = libc::sockaddr_in6 { + sin6_family: libc::AF_INET6 as libc::sa_family_t, + sin6_port: v6.port().to_be(), + sin6_flowinfo: v6.flowinfo(), + sin6_addr: libc::in6_addr { + s6_addr: v6.ip().octets(), + }, + sin6_scope_id: v6.scope_id(), + }; + ( + SockAddrUnion { v6: raw }, + std::mem::size_of::() as libc::socklen_t, + ) + } + } +} + +fn parse_addr(addr: &str) -> io::Result { + addr.parse().map_err(|_| { + io::Error::new( + io::ErrorKind::InvalidInput, + format!("{addr:?} is not a resolved ip:port — resolution is the c9 seam"), + ) + }) +} + +fn so_error(fd: RawFd) -> io::Result<()> { + let mut err: libc::c_int = 0; + let mut len = std::mem::size_of::() as libc::socklen_t; + let rc = unsafe { + libc::getsockopt( + fd, + libc::SOL_SOCKET, + libc::SO_ERROR, + (&mut err) as *mut _ as *mut libc::c_void, + &mut len, + ) + }; + if rc != 0 { + return Err(io::Error::last_os_error()); + } + if err != 0 { + return Err(io::Error::from_raw_os_error(err)); + } + Ok(()) +} + +// --------------------------------------------------------------------------- +// Conn +// --------------------------------------------------------------------------- + +/// One established TCP control connection. Owns the socket; drop closes it. +pub struct TcpConn { + stream: TcpStream, + closed: bool, +} + +impl TcpConn { + fn fd(&self) -> RawFd { + self.stream.as_raw_fd() + } +} + +impl Conn for TcpConn { + fn read(&mut self, buf: &mut [u8]) -> io::Result { + if self.closed { + return Ok(0); + } + if buf.is_empty() { + return Ok(0); + } + loop { + wait_readable(self.fd())?; + let n = unsafe { libc::read(self.fd(), buf.as_mut_ptr() as *mut _, buf.len()) }; + if n >= 0 { + return Ok(n as usize); + } + let e = io::Error::last_os_error(); + match e.kind() { + // Spurious readiness or signal: park again. + io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted => continue, + _ => return Err(e), + } + } + } + + fn write_all(&mut self, mut buf: &[u8]) -> io::Result<()> { + if self.closed { + return Err(io::Error::new( + io::ErrorKind::NotConnected, + "tcp conn closed locally", + )); + } + while !buf.is_empty() { + wait_writable(self.fd())?; + let n = unsafe { + libc::send( + self.fd(), + buf.as_ptr() as *const _, + buf.len(), + libc::MSG_NOSIGNAL, + ) + }; + if n >= 0 { + buf = &buf[n as usize..]; + continue; + } + let e = io::Error::last_os_error(); + match e.kind() { + io::ErrorKind::WouldBlock | io::ErrorKind::Interrupted => continue, + _ => return Err(e), + } + } + Ok(()) + } + + fn close(&mut self) { + if !self.closed { + self.closed = true; + // Best-effort: the peer sees EOF after draining. The fd itself + // is released when the owning stream drops. + let _ = self.stream.shutdown(std::net::Shutdown::Both); + } + } + + fn peer_addr(&self) -> String { + match self.stream.peer_addr() { + Ok(sa) => sa.to_string(), + Err(_) => "".to_string(), + } + } +} + +// --------------------------------------------------------------------------- +// Listener +// --------------------------------------------------------------------------- + +/// A bound TCP listen point (non-blocking socket; accept parks the actor). +pub struct TcpListener { + inner: StdListener, + local: SocketAddr, +} + +impl Listener for TcpListener { + fn accept(&mut self) -> io::Result> { + loop { + wait_readable(self.inner.as_raw_fd())?; + match self.inner.accept() { + Ok((stream, _peer)) => { + stream.set_nonblocking(true)?; + return Ok(Box::new(TcpConn { + stream, + closed: false, + })); + } + Err(e) + if e.kind() == io::ErrorKind::WouldBlock + || e.kind() == io::ErrorKind::Interrupted => + { + continue; + } + Err(e) => return Err(e), + } + } + } + + fn local_addr(&self) -> String { + self.local.to_string() + } +} + +// --------------------------------------------------------------------------- +// Transport +// --------------------------------------------------------------------------- + +/// The TCP transport. Stateless; every call stands alone. +pub struct TcpTransport; + +impl Transport for TcpTransport { + fn dial(&self, addr: &str) -> io::Result> { + let sa = parse_addr(addr)?; + let family = match sa { + SocketAddr::V4(_) => libc::AF_INET, + SocketAddr::V6(_) => libc::AF_INET6, + }; + let fd = unsafe { + libc::socket( + family, + libc::SOCK_STREAM | libc::SOCK_NONBLOCK | libc::SOCK_CLOEXEC, + 0, + ) + }; + if fd < 0 { + return Err(io::Error::last_os_error()); + } + // From here the fd is owned by `stream`; any early return drops it. + let stream = unsafe { + use std::os::fd::FromRawFd; + TcpStream::from_raw_fd(fd) + }; + let (raw, len) = to_sockaddr(&sa); + let rc = unsafe { libc::connect(fd, (&raw) as *const _ as *const libc::sockaddr, len) }; + if rc != 0 { + let e = io::Error::last_os_error(); + if e.raw_os_error() != Some(libc::EINPROGRESS) { + return Err(e); + } + // Connect in flight: park until the socket is writable, then the + // verdict is in SO_ERROR. + wait_writable(fd)?; + so_error(fd)?; + } + Ok(Box::new(TcpConn { + stream, + closed: false, + })) + } + + fn listen(&self, addr: &str) -> io::Result> { + let sa = parse_addr(addr)?; + let inner = StdListener::bind(sa)?; + inner.set_nonblocking(true)?; + let local = inner.local_addr()?; + Ok(Box::new(TcpListener { inner, local })) + } +} diff --git a/tests/cluster_transport.rs b/tests/cluster_transport.rs new file mode 100644 index 0000000..f9df0ae --- /dev/null +++ b/tests/cluster_transport.rs @@ -0,0 +1,272 @@ +//! RFC 010 c3 — transport conformance suite, run against both shipped impls +//! (TCP and in-memory loopback), plus impl-specific cases. +//! +//! Shared suite (roadmap): frame roundtrips through the framed codec, framing +//! across a split write, coalesced frames in one write, peer-close mid-frame +//! (must error, not EOF), clean close at a frame boundary (EOF as `Ok(None)`). +//! +//! The TCP impl parks the calling actor, so its runs live inside `smarm::run`; +//! loopback blocks the OS thread and runs as plain tests. +#![cfg(feature = "cluster")] + +use smarm::cluster::envelope::Frame; +use smarm::cluster::transport::loopback::LoopbackTransport; +use smarm::cluster::transport::tcp::TcpTransport; +use smarm::cluster::transport::{Conn, FramedConn, RecvError, Transport}; + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +/// Listener + dial + accept against one transport, both conns returned. +/// Relies on dial not requiring a concurrent accept (TCP backlog / loopback +/// queue), so a single thread or actor can hold both ends. +fn pair(t: &dyn Transport, addr: &str) -> (Box, Box) { + let mut l = t.listen(addr).unwrap(); + let a = t.dial(&l.local_addr()).unwrap(); + let b = l.accept().unwrap(); + (a, b) +} + +fn frames() -> Vec { + vec![ + Frame::Heartbeat, + Frame::Send { + index: 42, + generation: 3, + type_hash: 0x1234_5678_9ABC_DEF0, + payload: vec![1, 2, 3, 4, 5], + }, + Frame::SendNamed { + name: "the_counter".into(), + type_hash: 0xFFFF_0000_FFFF_0000, + payload: vec![], + }, + Frame::Demonitor { monitor_id: 77 }, + ] +} + +fn encode(f: &Frame) -> Vec { + let mut out = Vec::new(); + f.encode(&mut out).unwrap(); + out +} + +// --------------------------------------------------------------------------- +// Shared conformance suite — generic over an established pair +// --------------------------------------------------------------------------- + +fn suite_roundtrip(a: Box, b: Box) { + let mut fa = FramedConn::new(a); + let mut fb = FramedConn::new(b); + // a -> b, then b -> a: both directions carry every frame shape. + for f in frames() { + fa.send(&f).unwrap(); + assert_eq!(fb.recv().unwrap().unwrap(), f); + } + for f in frames() { + fb.send(&f).unwrap(); + assert_eq!(fa.recv().unwrap().unwrap(), f); + } +} + +fn suite_split_write(mut a: Box, b: Box) { + let f = Frame::Send { + index: 7, + generation: 1, + type_hash: 0xAB, + payload: vec![9; 64], + }; + let bytes = encode(&f); + // Split inside the length prefix, then inside the body: the reader must + // reassemble regardless of where the boundary falls. + a.write_all(&bytes[..2]).unwrap(); + a.write_all(&bytes[2..10]).unwrap(); + a.write_all(&bytes[10..]).unwrap(); + let mut fb = FramedConn::new(b); + assert_eq!(fb.recv().unwrap().unwrap(), f); +} + +fn suite_coalesced(mut a: Box, b: Box) { + let f1 = Frame::Heartbeat; + let f2 = Frame::Demonitor { monitor_id: 5 }; + let mut bytes = encode(&f1); + bytes.extend_from_slice(&encode(&f2)); + a.write_all(&bytes).unwrap(); + let mut fb = FramedConn::new(b); + assert_eq!(fb.recv().unwrap().unwrap(), f1); + assert_eq!(fb.recv().unwrap().unwrap(), f2); +} + +fn suite_close_mid_frame(mut a: Box, b: Box) { + let bytes = encode(&Frame::Send { + index: 1, + generation: 1, + type_hash: 1, + payload: vec![0; 128], + }); + a.write_all(&bytes[..bytes.len() / 2]).unwrap(); + a.close(); + let mut fb = FramedConn::new(b); + match fb.recv() { + Err(RecvError::TruncatedByPeer) => {} + other => panic!("expected TruncatedByPeer, got {other:?}"), + } +} + +fn suite_clean_close(mut a: Box, b: Box) { + let f = Frame::Heartbeat; + a.write_all(&encode(&f)).unwrap(); + a.close(); + let mut fb = FramedConn::new(b); + // The buffered frame is still delivered, then EOF at the boundary. + assert_eq!(fb.recv().unwrap().unwrap(), f); + assert!(fb.recv().unwrap().is_none()); +} + +fn run_suite(t: &dyn Transport, addr: &str) { + let (a, b) = pair(t, addr); + suite_roundtrip(a, b); + let (a, b) = pair(t, addr); + suite_split_write(a, b); + let (a, b) = pair(t, addr); + suite_coalesced(a, b); + let (a, b) = pair(t, addr); + suite_close_mid_frame(a, b); + let (a, b) = pair(t, addr); + suite_clean_close(a, b); +} + +// --------------------------------------------------------------------------- +// Loopback — plain tests, no runtime +// --------------------------------------------------------------------------- + +#[test] +fn loopback_conformance() { + // Fresh transport per pair() call is fine, but one instance must also + // support sequential re-listen on distinct addresses. + let t = LoopbackTransport::default(); + run_suite(&t, "alpha"); +} + +#[test] +fn loopback_dial_unknown_addr_refused() { + let t = LoopbackTransport::default(); + let err = t.dial("nobody-home").unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::ConnectionRefused); +} + +#[test] +fn loopback_addr_in_use() { + let t = LoopbackTransport::default(); + let _l = t.listen("alpha").unwrap(); + let err = t.listen("alpha").unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::AddrInUse); +} + +#[test] +fn loopback_listener_drop_frees_addr_and_refuses_dial() { + let t = LoopbackTransport::default(); + let l = t.listen("alpha").unwrap(); + drop(l); + let err = t.dial("alpha").unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::ConnectionRefused); + // Address is reusable after the listener is gone. + let _l2 = t.listen("alpha").unwrap(); +} + +#[test] +fn loopback_write_after_peer_close_broken_pipe() { + let t = LoopbackTransport::default(); + let (mut a, mut b) = pair(&t, "alpha"); + b.close(); + let err = a.write_all(&[1, 2, 3]).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::BrokenPipe); +} + +#[test] +fn loopback_cross_thread_blocking_read() { + // Reader blocks on an empty pipe until the writer thread delivers. + let t = LoopbackTransport::default(); + let (a, b) = pair(&t, "alpha"); + let mut fb = FramedConn::new(b); + let writer = std::thread::spawn(move || { + let mut a = a; + std::thread::sleep(std::time::Duration::from_millis(30)); + a.write_all(&encode(&Frame::Heartbeat)).unwrap(); + }); + assert_eq!(fb.recv().unwrap().unwrap(), Frame::Heartbeat); + writer.join().unwrap(); +} + +// --------------------------------------------------------------------------- +// TCP — inside the runtime (read/write park the calling actor) +// --------------------------------------------------------------------------- + +#[test] +fn tcp_conformance() { + smarm::run(|| { + run_suite(&TcpTransport, "127.0.0.1:0"); + }); +} + +#[test] +fn tcp_dial_refused() { + smarm::run(|| { + // Bind to an OS-assigned port, learn it, close the listener, dial it. + let addr = { + let l = TcpTransport.listen("127.0.0.1:0").unwrap(); + l.local_addr() + }; + let err = TcpTransport.dial(&addr).unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::ConnectionRefused); + }); +} + +#[test] +fn tcp_bad_addr_rejected_without_resolution() { + // Addresses are opaque pre-resolved strings; the c9 seam resolves names. + // A hostname is therefore invalid input here, not something to resolve. + let err = TcpTransport.dial("localhost:1234").unwrap_err(); + assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput); +} + +#[test] +fn tcp_local_addr_reports_real_port() { + let l = TcpTransport.listen("127.0.0.1:0").unwrap(); + let addr = l.local_addr(); + let port: u16 = addr.rsplit(':').next().unwrap().parse().unwrap(); + assert_ne!(port, 0); +} + +#[test] +fn tcp_big_frame_across_socket_buffers() { + // A payload far beyond socket buffer sizes forces genuine fragmentation + // and write backpressure: writer and reader must run concurrently. + smarm::run(|| { + let (tx, rx) = smarm::channel::(); + let payload = vec![0xA5u8; 4 * 1024 * 1024]; + let f = Frame::Send { + index: 9, + generation: 2, + type_hash: 0xC0FFEE, + payload, + }; + let mut l = TcpTransport.listen("127.0.0.1:0").unwrap(); + let addr = l.local_addr(); + let fw = f.clone(); + let writer = smarm::spawn(move || { + let mut fa = FramedConn::new(TcpTransport.dial(&addr).unwrap()); + fa.send(&fw).unwrap(); + }); + let reader = smarm::spawn(move || { + let mut fb = FramedConn::new(l.accept().unwrap()); + let got = fb.recv().unwrap().unwrap(); + tx.send(got).unwrap(); + }); + let got = rx.recv().unwrap(); + assert_eq!(got, f); + writer.join().unwrap(); + reader.join().unwrap(); + }); +}