feat(bus): tines-bus transport and daemon over a host IPC bus

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"

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]

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" }

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 <socket-path>`.
+
+fn main() {
+    let socket = match std::env::args().nth(1) {
+        Some(s) => s,
+        None => {
+            eprintln!("usage: tines-busd <socket-path>");
+            std::process::exit(2);
+        }
+    };
+    if let Err(e) = tines_bus::daemon::serve(&socket) {
+        eprintln!("tines-busd: {e}");
+        std::process::exit(1);
+    }
+}

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<u8> {
+    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<Option<Vec<u8>>> {
+    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<NodeId>,
+    stream: UnixStream,
+    buf: Vec<u8>,
+    /// 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<u8> },
+}
+
+/// 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<Path>) -> 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<Conn> = 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<Action> = 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<Action>) {
+    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<Conn>) {
+    // 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<NodeId> = 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<Vec<u8>> = 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);
+}

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};

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<FrameBuf>),
+    /// 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<BusEvent>;
+
+    /// 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<Vec<u8>>,
+    ev_rx: mpsc::Receiver<BusEvent>,
+}
+
+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<DaemonLink> {
+        let mut stream = retry_connect(socket)?;
+        let mut reader = stream.try_clone()?;
+
+        let (out_tx, out_rx) = mpsc::channel::<Vec<u8>>();
+        let (ev_tx, ev_rx) = mpsc::channel::<BusEvent>();
+
+        // 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<BusEvent> {
+        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<std::os::unix::net::UnixStream> {
+    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<L: BusLink = DaemonLink> {
+    me: NodeId,
+    allowlist: Option<NodeId>,
+    peers: Vec<NodeId>,
+    link: L,
+    closed: bool,
+}
+
+impl BusTransport<DaemonLink> {
+    /// 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<NodeId>,
+        socket: impl AsRef<Path>,
+    ) -> io::Result<BusTransport<DaemonLink>> {
+        let link = DaemonLink::connect(me, socket.as_ref())?;
+        Ok(BusTransport {
+            me,
+            allowlist,
+            peers: Vec::new(),
+            link,
+            closed: false,
+        })
+    }
+}
+
+impl<L: BusLink> BusTransport<L> {
+    /// 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<L: BusLink> Transport for BusTransport<L> {
+    type Addr = NodeId;
+
+    fn poll(&mut self) -> Option<Event<NodeId>> {
+        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()
+    }
+}

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");
+}

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<Option<Vec<u8>>> {
+    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]);
+}

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<u8> {
+    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<NodeId>) -> bool,
+    timeout: Duration,
+) -> Event<NodeId> {
+    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:?}"),
+    }
+}