From ee87808b059674b58da2f2dc3e8e239fe50dca37 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Sat, 24 Oct 2026 23:00:37 -0700 Subject: [PATCH] feat(bus): tines-bus transport and daemon over a host IPC bus --- Cargo.lock | 7 + Cargo.toml | 2 +- tines-bus/Cargo.toml | 15 ++ tines-bus/src/bin/tines-busd.rs | 16 ++ tines-bus/src/daemon.rs | 309 ++++++++++++++++++++++++++++++++ tines-bus/src/lib.rs | 13 ++ tines-bus/src/transport.rs | 285 +++++++++++++++++++++++++++++ tines-bus/tests/common/mod.rs | 34 ++++ tines-bus/tests/daemon.rs | 175 ++++++++++++++++++ tines-bus/tests/transport.rs | 181 +++++++++++++++++++ 10 files changed, 1036 insertions(+), 1 deletion(-) create mode 100644 tines-bus/Cargo.toml create mode 100644 tines-bus/src/bin/tines-busd.rs create mode 100644 tines-bus/src/daemon.rs create mode 100644 tines-bus/src/lib.rs create mode 100644 tines-bus/src/transport.rs create mode 100644 tines-bus/tests/common/mod.rs create mode 100644 tines-bus/tests/daemon.rs create mode 100644 tines-bus/tests/transport.rs diff --git a/Cargo.lock b/Cargo.lock index 4df26893..9ce9397d 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1484,6 +1484,13 @@ dependencies = [ "tokio", ] +[[package]] +name = "tines-bus" +version = "0.4.0" +dependencies = [ + "tines-core", +] + [[package]] name = "tines-core" version = "1.0.0" diff --git a/Cargo.toml b/Cargo.toml index a354bbcf..1db255b5 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["tines-core", "tines-sim", "tines-store", "tines-bt", "tines-ros2", "tines"] +members = ["tines-core", "tines-sim", "tines-store", "tines-bt", "tines-bus", "tines-ros2", "tines"] resolver = "2" [workspace.package] diff --git a/tines-bus/Cargo.toml b/tines-bus/Cargo.toml new file mode 100644 index 00000000..8ee67f18 --- /dev/null +++ b/tines-bus/Cargo.toml @@ -0,0 +1,15 @@ +[package] +name = "tines-bus" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "Bus transport for tines: exchange wire frames between nodes over a host IPC bus (Unix sockets), identity-preserving (Addr = NodeId)." +authors.workspace = true +repository.workspace = true +readme.workspace = true +documentation = "https://docs.rs/tines-bus" +keywords = ["robotics", "gossip", "delay-tolerant", "ipc", "unix-socket"] +categories = ["science::robotics", "network-programming"] + +[dependencies] +tines-core = { path = "../tines-core", version = "1" } diff --git a/tines-bus/src/bin/tines-busd.rs b/tines-bus/src/bin/tines-busd.rs new file mode 100644 index 00000000..eb7cfc4e --- /dev/null +++ b/tines-bus/src/bin/tines-busd.rs @@ -0,0 +1,16 @@ +//! The tines bus daemon: routes envelopes between connected clients on a +//! host IPC socket. Usage: `tines-busd `. + +fn main() { + let socket = match std::env::args().nth(1) { + Some(s) => s, + None => { + eprintln!("usage: tines-busd "); + std::process::exit(2); + } + }; + if let Err(e) = tines_bus::daemon::serve(&socket) { + eprintln!("tines-busd: {e}"); + std::process::exit(1); + } +} diff --git a/tines-bus/src/daemon.rs b/tines-bus/src/daemon.rs new file mode 100644 index 00000000..611d1cab --- /dev/null +++ b/tines-bus/src/daemon.rs @@ -0,0 +1,309 @@ +//! The bus daemon: a single-threaded, nonblocking hub that routes envelopes +//! between connected clients. +//! +//! # Protocol +//! +//! Every message is `[u32 LE length][payload]`, the length counting payload +//! bytes only (bus-local; the core frames inside are untouched). +//! +//! - **Hello** (client→daemon, first message): payload = 16-byte `NodeId`. +//! A hello for an already-registered id replaces the old connection (the +//! old socket is closed). +//! - **Data** (client→daemon): payload = `[16B from][16B to][frame]`. The +//! daemon drops (and counts on stderr) any message whose `from` is not that +//! socket's registered id, or whose `to` is not registered; otherwise it +//! forwards the envelope verbatim to `to`'s socket. Frames longer than +//! [`MAX_FRAME`] close the connection. +//! - **Control** (daemon→client): payload = `[1B kind][16B NodeId]`; kind +//! `0 = joined`, `1 = left`; sent to every other registered client on join, +//! and to all remaining on leave. +//! +//! EOF or an error from a client's socket is a leave plus close; the daemon +//! itself stays up. No frame content is decoded beyond the routing header. + +use std::io::{self, ErrorKind, Read, Write}; +use std::os::unix::net::{UnixListener, UnixStream}; +use std::path::Path; +use std::time::Duration; + +use tines_core::consts::MAX_FRAME; +use tines_core::NodeId; + +/// Routing-header size (from + to) of a data envelope. +const HEADER_BYTES: usize = 32; + +/// Longest envelope payload the daemon will buffer or forward. +const MAX_PAYLOAD: usize = MAX_FRAME + HEADER_BYTES; + +fn id_of(bytes: &[u8]) -> NodeId { + let mut id = [0u8; 16]; + id.copy_from_slice(bytes); + id +} + +fn control(kind: u8, id: NodeId) -> Vec { + let mut env = Vec::with_capacity(4 + 17); + env.extend_from_slice(&17u32.to_le_bytes()); + env.push(kind); + env.extend_from_slice(&id); + env +} + +/// Read one envelope's payload from `s`; `Ok(None)` means clean EOF. +pub(crate) fn read_envelope(s: &mut UnixStream) -> io::Result>> { + let mut len_buf = [0u8; 4]; + let n = s.read(&mut len_buf)?; + if n == 0 { + return Ok(None); + } + if n != 4 { + return Err(io::Error::new( + ErrorKind::InvalidData, + "short envelope length header", + )); + } + let len = u32::from_le_bytes(len_buf) as usize; + if len > MAX_PAYLOAD { + return Err(io::Error::new( + ErrorKind::InvalidData, + "envelope over the size cap", + )); + } + let mut payload = vec![0u8; len]; + s.read_exact(&mut payload)?; + Ok(Some(payload)) +} + +struct Conn { + id: Option, + stream: UnixStream, + buf: Vec, + /// Set to evict this connection on the next apply pass. + close_pending: bool, +} + +/// A queued action from processing one message. +enum Action { + /// Close the connection at `idx`; if it was registered, announce a leave. + Close { idx: usize }, + /// Write a prepared envelope to the connection at `to`. + Forward { to: usize, env: Vec }, +} + +/// Serve the bus on `socket` until the process dies. +/// +/// A stale socket file from a previous run is removed first so `bind` cannot +/// fail on it. +pub fn serve(socket: impl AsRef) -> io::Result<()> { + let socket = socket.as_ref().to_path_buf(); + let _ = std::fs::remove_file(&socket); + let listener = UnixListener::bind(&socket)?; + listener.set_nonblocking(true)?; + + let mut conns: Vec = Vec::new(); + loop { + // Accept everything pending. + loop { + match listener.accept() { + Ok((stream, _)) => { + stream.set_nonblocking(true)?; + conns.push(Conn { + id: None, + stream, + buf: Vec::new(), + close_pending: false, + }); + } + Err(ref e) if e.kind() == ErrorKind::WouldBlock => break, + Err(_) => break, + } + } + + // Read + parse each connection, queueing actions. + let mut actions: Vec = Vec::new(); + for i in 0..conns.len() { + let (closed, msgs) = { + let c = &mut conns[i]; + let mut closed = false; + let mut chunk = [0u8; 4096]; + loop { + match c.stream.read(&mut chunk) { + Ok(0) => { + closed = true; + break; + } + Ok(n) => { + c.buf.extend_from_slice(&chunk[..n]); + if n < chunk.len() { + break; + } + } + Err(ref e) if e.kind() == ErrorKind::WouldBlock => break, + Err(_) => { + closed = true; + break; + } + } + } + // Split complete envelopes out of the buffer. + let mut msgs = Vec::new(); + loop { + if c.buf.len() < 4 { + break; + } + let len = u32::from_le_bytes([c.buf[0], c.buf[1], c.buf[2], c.buf[3]]) as usize; + if len > MAX_PAYLOAD { + // Oversized envelope: protocol violation. + closed = true; + break; + } + if c.buf.len() < 4 + len { + break; + } + msgs.push(c.buf[4..4 + len].to_vec()); + c.buf.drain(..4 + len); + } + (closed, msgs) + }; + // Process complete messages even when the connection is being + // closed: a burst hello+oversized-data must still register the + // id so the leave is announced to the right peers. + for msg in msgs { + process_msg(i, &msg, &mut conns, &mut actions); + } + if closed { + actions.push(Action::Close { idx: i }); + } + } + + apply_actions(&mut actions, &mut conns); + + std::thread::sleep(Duration::from_millis(2)); + } +} + +/// Handle one envelope from the connection at `idx`. +fn process_msg(idx: usize, payload: &[u8], conns: &mut [Conn], actions: &mut Vec) { + let (before, rest) = conns.split_at_mut(idx); + let (me, after) = rest.split_at_mut(1); + let me = &mut me[0]; + + match me.id { + None => { + // Unregistered sockets may only hello. + if payload.len() != 16 { + actions.push(Action::Close { idx }); + return; + } + let id = id_of(payload); + // A hello for an already-registered id replaces the old + // connection (silently: peers never see a flap). + let is_new = !before.iter().chain(after.iter()).any(|c| c.id == Some(id)); + if !is_new { + for c in before.iter_mut().chain(after.iter_mut()) { + if c.id == Some(id) { + c.close_pending = true; + } + } + } + me.id = Some(id); + if is_new { + let env = control(0, id); + for (j, c) in before + .iter() + .enumerate() + .chain(after.iter().enumerate().map(|(k, c)| (k + idx + 1, c))) + { + if c.id.is_some() { + actions.push(Action::Forward { + to: j, + env: env.clone(), + }); + } + } + } + } + Some(my_id) => { + if payload.len() == 16 { + // Repeat hello on an already-registered socket: no-op. + return; + } + if payload.len() < 32 { + actions.push(Action::Close { idx }); + return; + } + let from = id_of(&payload[0..16]); + let to = id_of(&payload[16..32]); + let frame_len = payload.len() - 32; + if from != my_id { + // The daemon certifies `from`: never route a forgery. + eprintln!("tines-busd: dropped: from mismatch on registered socket"); + return; + } + if frame_len > MAX_FRAME { + actions.push(Action::Close { idx }); + return; + } + // `position()` on the chain is the actual index only for peers + // before the sender; after the sender it is one short. + let to_idx = before.iter().position(|c| c.id == Some(to)).or_else(|| { + after + .iter() + .position(|c| c.id == Some(to)) + .map(|k| k + idx + 1) + }); + match to_idx { + Some(j) => { + let mut env = Vec::with_capacity(4 + payload.len()); + env.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + env.extend_from_slice(payload); + actions.push(Action::Forward { to: j, env }); + } + None => eprintln!("tines-busd: dropped: unknown destination"), + } + } + } +} + +/// Run queued actions: forwards first, then closes (announcing leaves). +fn apply_actions(actions: &mut [Action], conns: &mut Vec) { + // Forwards (only to live, registered connections). + for a in actions.iter() { + if let Action::Forward { to, env } = a { + if let Some(c) = conns.get_mut(*to) { + if !c.close_pending && c.id.is_some() && c.stream.write_all(env).is_err() { + c.close_pending = true; + } + } + } + } + + // Closes. A registered connection that leaves (EOF, error, protocol + // violation) is announced to everyone remaining; an evicted duplicate + // already has `close_pending` set and is silent. + let mut announced: Vec = Vec::new(); + for a in actions.iter() { + if let Action::Close { idx } = a { + if let Some(c) = conns.get_mut(*idx) { + c.close_pending = true; + if let Some(id) = c.id { + announced.push(id); + } + } + } + } + if !announced.is_empty() { + let envs: Vec> = announced.iter().map(|id| control(1, *id)).collect(); + for env in envs { + for c in conns.iter_mut() { + if c.close_pending || c.id.is_none() { + continue; + } + if c.stream.write_all(&env).is_err() { + c.close_pending = true; + } + } + } + } + conns.retain(|c| !c.close_pending); +} diff --git a/tines-bus/src/lib.rs b/tines-bus/src/lib.rs new file mode 100644 index 00000000..fc7e6992 --- /dev/null +++ b/tines-bus/src/lib.rs @@ -0,0 +1,13 @@ +//! The tines bus transport: exchange wire frames between nodes over a host +//! IPC bus (Unix sockets), identity-preserving (`Addr = NodeId`). +//! +//! [`BusTransport`] is the tines [`Transport`] implementation; [`daemon`] is +//! the hub that routes frames between all connected nodes. The two sides of +//! a fleet talk through one shared daemon socket. + +#![deny(missing_docs)] + +pub mod daemon; +mod transport; + +pub use transport::{BusError, BusEvent, BusLink, BusTransport, DaemonLink}; diff --git a/tines-bus/src/transport.rs b/tines-bus/src/transport.rs new file mode 100644 index 00000000..10bb8726 --- /dev/null +++ b/tines-bus/src/transport.rs @@ -0,0 +1,285 @@ +//! The bus transport: a [`Transport`] whose peers are other tines nodes +//! connected to the same host bus daemon (see [`crate::daemon`]). +//! +//! The pure state machine ([`BusTransport`]) is generic over a narrow I/O +//! seam, [`BusLink`], mirroring how `tines-bt` splits logic from radio. The +//! real link ([`DaemonLink`]) is a thread-driven client of the bus daemon; +//! it lives in this crate but is not part of the pure logic. + +use std::io::{self, Write}; +use std::path::Path; +use std::sync::mpsc; + +use tines_core::consts::MAX_FRAME; +use tines_core::{frame_sender, Event, FrameBuf, SendError, Transport}; + +use crate::daemon; +pub use tines_core::NodeId; + +/// An event observed on the bus link. +#[derive(Debug)] +pub enum BusEvent { + /// A peer registered with the daemon. + Joined(NodeId), + /// A registered peer left (or its connection was replaced). + Left(NodeId), + /// A data envelope arrived, attributed to the daemon-certified sender. + Frame(NodeId, Box), + /// The link itself closed (daemon gone, socket error, or local shutdown). + Closed, +} + +/// Errors a [`BusLink`] can report from `send`. +#[derive(Debug)] +pub enum BusError { + /// The underlying socket/channel failed. + Io(io::Error), + /// The destination is not reachable on this link. + Unreachable, +} + +/// The narrow I/O seam behind [`BusTransport`]. +/// +/// `send` receives the *frame bytes only*; the link stamps the envelope with +/// the sender's identity, so a caller cannot forge who a frame is from. +pub trait BusLink { + /// Poll for the next link event, if any is ready. + fn poll(&mut self) -> Option; + + /// Queue `data` for delivery to `to`. + fn send(&mut self, to: &NodeId, data: &[u8]) -> Result<(), BusError>; + + /// Largest frame this link can carry (its transport MTU). + fn mtu(&self) -> usize; +} + +/// A thread-driven client of the bus daemon. +/// +/// A writer thread owns the socket (sending the hello first, then queued +/// envelopes) and a reader thread owns a clone, decoding inbound envelopes +/// into [`BusEvent`]s. Dropping the link drops the writer's queue, which +/// shuts the socket down, which unblocks the reader — no explicit shutdown +/// protocol needed. +#[derive(Debug)] +pub struct DaemonLink { + me: NodeId, + out_tx: mpsc::Sender>, + ev_rx: mpsc::Receiver, +} + +impl DaemonLink { + /// Connect to the daemon at `socket` and hello as `me`. + /// + /// The connect retries briefly because a freshly spawned daemon binds + /// asynchronously. + pub fn connect(me: NodeId, socket: &Path) -> io::Result { + let mut stream = retry_connect(socket)?; + let mut reader = stream.try_clone()?; + + let (out_tx, out_rx) = mpsc::channel::>(); + let (ev_tx, ev_rx) = mpsc::channel::(); + + // Writer thread: hello first, then queued envelopes. On queue + // disconnect (the transport was dropped) it shuts the socket down, + // which also unblocks the reader thread. + let mut hello = Vec::with_capacity(4 + 16); + hello.extend_from_slice(&16u32.to_le_bytes()); + hello.extend_from_slice(&me); + std::thread::spawn(move || { + if stream.write_all(&hello).is_err() { + let _ = stream.shutdown(std::net::Shutdown::Both); + return; + } + while let Ok(env) = out_rx.recv() { + if stream.write_all(&env).is_err() { + break; + } + } + let _ = stream.shutdown(std::net::Shutdown::Both); + }); + + // Reader thread: decode envelopes into events. + std::thread::spawn(move || { + loop { + let payload = match daemon::read_envelope(&mut reader) { + Ok(Some(p)) => p, + Ok(None) | Err(_) => { + let _ = ev_tx.send(BusEvent::Closed); + return; + } + }; + let ev = match payload.len() { + 17 => match payload[0] { + 0 => BusEvent::Joined(id_of(&payload[1..17])), + _ => BusEvent::Left(id_of(&payload[1..17])), + }, + n if n >= 32 => { + let from = id_of(&payload[0..16]); + let frame = match FrameBuf::from_slice(&payload[32..]) { + Ok(f) => f, + // Overlong frame: protocol violation, treat as + // a dead link. + Err(_) => { + let _ = ev_tx.send(BusEvent::Closed); + return; + } + }; + BusEvent::Frame(from, Box::new(frame)) + } + _ => continue, // unknown message shape: ignore + }; + if ev_tx.send(ev).is_err() { + return; + } + } + }); + + Ok(DaemonLink { me, out_tx, ev_rx }) + } +} + +impl BusLink for DaemonLink { + fn poll(&mut self) -> Option { + self.ev_rx.try_recv().ok() + } + + fn send(&mut self, to: &NodeId, data: &[u8]) -> Result<(), BusError> { + // Envelope: [u32 LE payload len][from][to][frame]. + let mut env = Vec::with_capacity(4 + 32 + data.len()); + env.extend_from_slice(&((32 + data.len()) as u32).to_le_bytes()); + env.extend_from_slice(&self.me); + env.extend_from_slice(to); + env.extend_from_slice(data); + self.out_tx + .send(env) + .map_err(|_| BusError::Io(io::Error::new(io::ErrorKind::BrokenPipe, "bus link closed"))) + } + + fn mtu(&self) -> usize { + MAX_FRAME + } +} + +/// Connect to `socket`, retrying briefly while a just-spawned daemon binds. +fn retry_connect(socket: &Path) -> io::Result { + let mut last_err = None; + for _ in 0..40 { + match std::os::unix::net::UnixStream::connect(socket) { + Ok(s) => return Ok(s), + Err(e) => last_err = Some(e), + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + Err(last_err.unwrap_or_else(|| { + io::Error::new(io::ErrorKind::ConnectionRefused, "bus daemon never bound") + })) +} + +fn id_of(bytes: &[u8]) -> NodeId { + let mut id = [0u8; 16]; + id.copy_from_slice(bytes); + id +} + +/// The bus transport: tines `Transport` over a [`BusLink`]. +/// +/// `type Addr = NodeId` — the daemon certifies each message's sender, so bus +/// peer identity is as strong as the other transports'. +pub struct BusTransport { + me: NodeId, + allowlist: Option, + peers: Vec, + link: L, + closed: bool, +} + +impl BusTransport { + /// Create a bus transport connected to the daemon at `socket`. + /// + /// `allowlist` pins the one peer this transport may see and reach + /// (`None` = every bus peer), mirroring the single-peer allowlist of the + /// bt transport. + pub fn new( + me: NodeId, + allowlist: Option, + socket: impl AsRef, + ) -> io::Result> { + let link = DaemonLink::connect(me, socket.as_ref())?; + Ok(BusTransport { + me, + allowlist, + peers: Vec::new(), + link, + closed: false, + }) + } +} + +impl BusTransport { + /// Whether `peer` passes the allowlist (and is not ourselves). + fn allows(&self, peer: &NodeId) -> bool { + *peer != self.me && self.allowlist.is_none_or(|a| a == *peer) + } +} + +impl Transport for BusTransport { + type Addr = NodeId; + + fn poll(&mut self) -> Option> { + loop { + match self.link.poll()? { + BusEvent::Joined(addr) => { + if !self.allows(&addr) || self.peers.contains(&addr) { + continue; + } + self.peers.push(addr); + return Some(Event::Contact(addr)); + } + BusEvent::Left(addr) => { + if let Some(pos) = self.peers.iter().position(|p| *p == addr) { + self.peers.remove(pos); + return Some(Event::Lost(addr)); + } + } + BusEvent::Frame(peer, frame) => { + if !self.allows(&peer) { + continue; + } + // The daemon certifies the envelope's `from`; the frame's + // own declared sender must agree with it. + if frame_sender(frame.as_slice()) != Some(peer) { + continue; + } + return Some(Event::Frame(peer, *frame)); + } + BusEvent::Closed => { + if self.closed { + return None; + } + self.closed = true; + // Every bound peer is gone; drain one `Lost` per poll. + return self.peers.pop().map(Event::Lost); + } + } + } + } + + fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> { + if !self.allows(to) || !self.peers.contains(to) { + return Err(SendError::Unreachable); + } + if frame.len() > self.mtu() { + return Err(SendError::TooLarge); + } + match self.link.send(to, frame) { + Ok(()) => Ok(()), + // The only failure modes left are a dead link (nothing is + // reachable any more) and a bad destination. + Err(_) => Err(SendError::Unreachable), + } + } + + fn mtu(&self) -> usize { + self.link.mtu() + } +} diff --git a/tines-bus/tests/common/mod.rs b/tines-bus/tests/common/mod.rs new file mode 100644 index 00000000..bcafdc35 --- /dev/null +++ b/tines-bus/tests/common/mod.rs @@ -0,0 +1,34 @@ +//! Shared helpers for the tines-bus integration tests. +#![allow(dead_code)] + +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicUsize, Ordering}; + +static COUNTER: AtomicUsize = AtomicUsize::new(0); + +/// A fresh, unique scratch directory under the system temp dir. +pub fn unique_dir(tag: &str) -> PathBuf { + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let dir = std::env::temp_dir().join(format!("tines-bus-{tag}-{}-{n}", std::process::id())); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +/// Spawn the bus daemon on a thread and return its socket path once bound. +pub fn spawn_daemon(dir: &Path) -> PathBuf { + let sock = dir.join("bus.sock"); + let s = sock.clone(); + std::thread::spawn(move || { + if let Err(e) = tines_bus::daemon::serve(&s) { + panic!("daemon failed: {e}"); + } + }); + // Wait for the daemon to bind, so the first connect cannot race it. + for _ in 0..400 { + if sock.exists() { + return sock; + } + std::thread::sleep(std::time::Duration::from_millis(5)); + } + panic!("daemon did not bind in time"); +} diff --git a/tines-bus/tests/daemon.rs b/tines-bus/tests/daemon.rs new file mode 100644 index 00000000..49cc2764 --- /dev/null +++ b/tines-bus/tests/daemon.rs @@ -0,0 +1,175 @@ +//! Protocol-level tests driving raw Unix sockets straight at the daemon. +mod common; + +use common::{spawn_daemon, unique_dir}; +use std::io::{ErrorKind, Read, Write}; +use std::os::unix::net::UnixStream; +use std::time::{Duration, Instant}; +use tines_core::consts::MAX_FRAME; + +type NodeId = [u8; 16]; + +fn connect(sock: &std::path::Path) -> UnixStream { + let s = UnixStream::connect(sock).unwrap(); + s.set_read_timeout(Some(Duration::from_secs(2))).unwrap(); + s +} + +fn send_envelope(s: &mut UnixStream, payload: &[u8]) { + let mut buf = Vec::with_capacity(4 + payload.len()); + buf.extend_from_slice(&(payload.len() as u32).to_le_bytes()); + buf.extend_from_slice(payload); + s.write_all(&buf).unwrap(); +} + +fn hello(s: &mut UnixStream, id: &NodeId) { + send_envelope(s, id); +} + +fn data(s: &mut UnixStream, from: &NodeId, to: &NodeId, frame: &[u8]) { + let mut p = Vec::with_capacity(32 + frame.len()); + p.extend_from_slice(from); + p.extend_from_slice(to); + p.extend_from_slice(frame); + send_envelope(s, &p); +} + +/// Read one envelope's payload; `Ok(None)` means EOF. +fn read_envelope(s: &mut UnixStream) -> std::io::Result>> { + let mut len_buf = [0u8; 4]; + let n = s.read(&mut len_buf)?; + if n == 0 { + return Ok(None); + } + assert_eq!(n, 4, "short length header"); + let len = u32::from_le_bytes(len_buf) as usize; + let mut payload = vec![0u8; len]; + s.read_exact(&mut payload)?; + Ok(Some(payload)) +} + +fn expect_control(s: &mut UnixStream, kind: u8, id: &NodeId) { + let p = read_envelope(s) + .unwrap() + .expect("connection closed unexpectedly"); + assert_eq!(p.len(), 17, "control payload is kind + 16-byte id"); + assert_eq!(p[0], kind); + assert_eq!(&p[1..17], id); +} + +/// Assert the socket stays silent for `window_ms`, then return. +fn expect_no_data(s: &mut UnixStream, window_ms: u64) { + s.set_read_timeout(Some(Duration::from_millis(100))) + .unwrap(); + let deadline = Instant::now() + Duration::from_millis(window_ms); + let mut buf = [0u8; 1]; + loop { + match s.read(&mut buf) { + Ok(0) => panic!("connection closed, expected silence"), + Ok(_) => panic!("expected no data, got some"), + Err(e) if e.kind() == ErrorKind::WouldBlock || e.kind() == ErrorKind::TimedOut => { + if Instant::now() >= deadline { + return; + } + std::thread::sleep(Duration::from_millis(5)); + } + Err(e) => panic!("unexpected read error: {e}"), + } + } +} + +#[test] +fn duplicate_hello_replaces() { + let dir = unique_dir("dup"); + let sock = spawn_daemon(&dir); + + let mut s1 = connect(&sock); + hello(&mut s1, &[1; 16]); + let mut s2 = connect(&sock); + hello(&mut s2, &[1; 16]); + + // The first connection is closed by the daemon... + let mut buf = [0u8; 1]; + assert_eq!( + s1.read(&mut buf).unwrap(), + 0, + "first connection must see EOF after replacement" + ); + + // ...and the daemon stays up, serving the replacement: a third peer's + // join reaches s2 as a control message. + let mut s3 = connect(&sock); + hello(&mut s3, &[2; 16]); + expect_control(&mut s2, 0, &[2; 16]); +} + +#[test] +fn mismatched_from_is_dropped() { + let dir = unique_dir("mismatch"); + let sock = spawn_daemon(&dir); + + let mut a = connect(&sock); + hello(&mut a, &[1; 16]); + let mut b = connect(&sock); + hello(&mut b, &[2; 16]); + + // A (registered as [1;16]) claims to be [2;16]: nothing may reach B. + data(&mut a, &[2; 16], &[2; 16], &[0xAB; 10]); + expect_no_data(&mut b, 300); + + // The daemon survives and A is still connected: a well-formed message + // is forwarded verbatim. + data(&mut a, &[1; 16], &[2; 16], &[0xCD; 10]); + let p = read_envelope(&mut b).unwrap().expect("connection closed"); + assert_eq!(p.len(), 42); + assert_eq!(&p[0..16], &[1; 16]); + assert_eq!(&p[16..32], &[2; 16]); + assert_eq!(&p[32..42], &[0xCD; 10]); +} + +#[test] +fn unknown_dest_is_dropped() { + let dir = unique_dir("unknown"); + let sock = spawn_daemon(&dir); + + let mut a = connect(&sock); + hello(&mut a, &[1; 16]); + + // No [9;16] registered: dropped, and A stays connected. + data(&mut a, &[1; 16], &[9; 16], &[0xAB; 10]); + + let mut b = connect(&sock); + hello(&mut b, &[2; 16]); + // Wait until the daemon has registered B: A sees the join control. + expect_control(&mut a, 0, &[2; 16]); + data(&mut a, &[1; 16], &[2; 16], &[0xCD; 10]); + let p = read_envelope(&mut b).unwrap().expect("connection closed"); + assert_eq!(p.len(), 42); + assert_eq!(&p[32..42], &[0xCD; 10]); +} + +#[test] +fn oversized_frame_closes_connection() { + let dir = unique_dir("oversize"); + let sock = spawn_daemon(&dir); + + let mut a = connect(&sock); + hello(&mut a, &[1; 16]); + let mut b = connect(&sock); + hello(&mut b, &[2; 16]); + + // Handshake: once A has seen B join, both hellos are registered. + expect_control(&mut a, 0, &[2; 16]); + + // A frame one byte over MAX_FRAME: that connection is closed... + data(&mut a, &[1; 16], &[2; 16], &vec![0u8; MAX_FRAME + 1]); + let mut buf = [0u8; 1]; + assert_eq!( + a.read(&mut buf).unwrap(), + 0, + "oversized frame must close the connection" + ); + + // ...and B learns A left; the daemon stays up for everyone else. + expect_control(&mut b, 1, &[1; 16]); +} diff --git a/tines-bus/tests/transport.rs b/tines-bus/tests/transport.rs new file mode 100644 index 00000000..b5203eb1 --- /dev/null +++ b/tines-bus/tests/transport.rs @@ -0,0 +1,181 @@ +//! End-to-end tests for the bus transport: real `BusTransport`s on both ends +//! of a live daemon, exchanging core wire frames. +mod common; + +use common::{spawn_daemon, unique_dir}; +use std::path::Path; +use std::time::{Duration, Instant}; +use tines_bus::BusTransport; +use tines_core::{consts, Event, SendError, Transport}; + +type NodeId = [u8; 16]; + +/// A minimally valid wire frame declaring `from` as its sender: +/// `WIRE_VERSION`, a variant tag byte, then the 16-byte sender id +/// (`frame_sender` reads exactly those bytes). +fn frame_from(from: &NodeId) -> Vec { + let mut f = Vec::with_capacity(18); + f.push(consts::WIRE_VERSION); + f.push(0); + f.extend_from_slice(from); + f +} + +/// Poll `t` until `pred` accepts an event, or panic after `timeout`. +fn poll_until( + t: &mut BusTransport, + mut pred: impl FnMut(&Event) -> bool, + timeout: Duration, +) -> Event { + let deadline = Instant::now() + timeout; + loop { + if let Some(ev) = t.poll() { + if pred(&ev) { + return ev; + } + } + assert!( + Instant::now() < deadline, + "timed out waiting for a bus event" + ); + std::thread::sleep(Duration::from_millis(5)); + } +} + +/// Assert `t` yields no events for `window`, then give up. +fn expect_quiet(t: &mut BusTransport, window: Duration) { + let deadline = Instant::now() + window; + while Instant::now() < deadline { + assert!(t.poll().is_none(), "expected no events, got one"); + std::thread::sleep(Duration::from_millis(5)); + } +} + +fn daemon_for(dir: &Path) -> std::path::PathBuf { + spawn_daemon(dir) +} + +#[test] +fn two_transports_exchange_frames() { + let dir = unique_dir("xfer"); + let sock = daemon_for(&dir); + let mut a = BusTransport::new([1; 16], None, &sock).unwrap(); + let mut b = BusTransport::new([2; 16], None, &sock).unwrap(); + + // A learns about B when B joins. + match poll_until( + &mut a, + |ev| matches!(ev, Event::Contact(_)), + Duration::from_secs(3), + ) { + Event::Contact(id) => assert_eq!(id, [2; 16]), + other => panic!("expected Contact, got {other:?}"), + } + + let frame = frame_from(&[1; 16]); + a.send(&[2; 16], &frame).unwrap(); + + // B receives the frame from A with identical bytes. + match poll_until( + &mut b, + |ev| matches!(ev, Event::Frame(_, _)), + Duration::from_secs(3), + ) { + Event::Frame(peer, fbuf) => { + assert_eq!(peer, [1; 16]); + assert_eq!(fbuf.as_slice(), frame.as_slice()); + } + other => panic!("expected Frame, got {other:?}"), + } +} + +#[test] +fn spoofed_frame_is_dropped() { + let dir = unique_dir("spoof"); + let sock = daemon_for(&dir); + let mut a = BusTransport::new([1; 16], None, &sock).unwrap(); + let mut b = BusTransport::new([2; 16], None, &sock).unwrap(); + + poll_until( + &mut a, + |ev| matches!(ev, Event::Contact(_)), + Duration::from_secs(3), + ); + + // A (complicit) sends a frame that claims to be from C: the bus stamps + // A as the envelope sender, so B must never surface it. + let spoofed = frame_from(&[9; 16]); + a.send(&[2; 16], &spoofed).unwrap(); + expect_quiet(&mut b, Duration::from_millis(400)); + + // The link is still healthy: a genuine frame from A arrives intact. + let real = frame_from(&[1; 16]); + a.send(&[2; 16], &real).unwrap(); + match poll_until( + &mut b, + |ev| matches!(ev, Event::Frame(_, _)), + Duration::from_secs(3), + ) { + Event::Frame(peer, fbuf) => { + assert_eq!(peer, [1; 16]); + assert_eq!(fbuf.as_slice(), real.as_slice()); + } + other => panic!("expected Frame, got {other:?}"), + } +} + +#[test] +fn join_and_leave_become_contact_and_lost() { + let dir = unique_dir("joinleave"); + let sock = daemon_for(&dir); + let mut a = BusTransport::new([1; 16], None, &sock).unwrap(); + let b = BusTransport::new([2; 16], None, &sock).unwrap(); + + match poll_until( + &mut a, + |ev| matches!(ev, Event::Contact(_)), + Duration::from_secs(3), + ) { + Event::Contact(id) => assert_eq!(id, [2; 16]), + other => panic!("expected Contact, got {other:?}"), + } + + drop(b); + match poll_until( + &mut a, + |ev| matches!(ev, Event::Lost(_)), + Duration::from_secs(3), + ) { + Event::Lost(id) => assert_eq!(id, [2; 16]), + other => panic!("expected Lost, got {other:?}"), + } +} + +#[test] +fn allowlist_filters() { + let dir = unique_dir("allow"); + let sock = daemon_for(&dir); + let mut a = BusTransport::new([1; 16], Some([2; 16]), &sock).unwrap(); + let _b = BusTransport::new([2; 16], None, &sock).unwrap(); + let _c = BusTransport::new([3; 16], None, &sock).unwrap(); + + // Only the allowlisted peer surfaces. + match poll_until( + &mut a, + |ev| matches!(ev, Event::Contact(_)), + Duration::from_secs(3), + ) { + Event::Contact(id) => assert_eq!(id, [2; 16]), + other => panic!("expected Contact, got {other:?}"), + } + + // C's join is filtered out: nothing else surfaces. + expect_quiet(&mut a, Duration::from_millis(400)); + + // And sending to a non-allowlisted peer is refused. + let frame = frame_from(&[1; 16]); + match a.send(&[3; 16], &frame) { + Err(SendError::Unreachable) => {} + other => panic!("expected Unreachable, got {other:?}"), + } +}