diff --git a/Cargo.lock b/Cargo.lock index 9b369670..24c22af5 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1460,6 +1460,18 @@ dependencies = [ "syn 3.0.9", ] +[[package]] +name = "tines" +version = "0.3.0" +dependencies = [ + "heapless 0.8.0", + "serde", + "serde_json", + "tines-bt", + "tines-core", + "tines-store", +] + [[package]] name = "tines-bt" version = "0.3.0" diff --git a/Cargo.toml b/Cargo.toml index 4941bf13..cc7c0a56 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,5 +1,5 @@ [workspace] -members = ["tines-core", "tines-sim", "tines-store", "tines-bt", "tines-ros2"] +members = ["tines-core", "tines-sim", "tines-store", "tines-bt", "tines-ros2", "tines"] resolver = "2" [workspace.package] diff --git a/tines/Cargo.toml b/tines/Cargo.toml new file mode 100644 index 00000000..ed10ce4e --- /dev/null +++ b/tines/Cargo.toml @@ -0,0 +1,25 @@ +[package] +name = "tines" +version.workspace = true +edition.workspace = true +license.workspace = true +description = "Standalone tines service: hosts a tines Node over a Unix-domain-socket IPC for robots without ROS2." +authors.workspace = true +repository.workspace = true +readme.workspace = true +keywords = ["robotics", "delay-tolerant", "science::robotics", "ipc", "ble"] +categories = ["science::robotics", "network-programming"] + +[dependencies] +tines-core = { path = "../tines-core", version = "0.3.0" } +tines-store = { path = "../tines-store", version = "0.3.0" } +tines-bt = { path = "../tines-bt", version = "0.3.0" } +serde = { version = "1", features = ["derive"] } +serde_json = "1" + +[dev-dependencies] +tines-core = { path = "../tines-core", version = "0.3.0" } +tines-store = { path = "../tines-store", version = "0.3.0" } +heapless = "0.8" +serde = { version = "1", features = ["derive"] } +serde_json = "1" diff --git a/tines/src/ipc.rs b/tines/src/ipc.rs new file mode 100644 index 00000000..499ec64c --- /dev/null +++ b/tines/src/ipc.rs @@ -0,0 +1,297 @@ +//! IPC over a Unix-domain socket. +//! +//! The [`serve`] fn hosts a tines [`Node`] and speaks a tiny +//! request/response protocol to each connecting peer. A single instance is the +//! control plane for its device; external tools (or another tines Node, in +//! later tasks) drive the node through four operations: `observe`, `belief`, +//! `events`, and `status`. +//! +//! # Wire format +//! +//! Length-prefixed JSON, one message per direction: +//! +//! ```text +//! [u32 : little-endian byte length] [u8 * length : a JSON frame] +//! ``` +//! +//! Each frame is one of the [`Request`] or [`Response`] enums, tagged by a +//! `"kind"` field (serde externally-tagged). Frames are capped at 1 MiB; a +//! longer frame is treated as a malformed request and the connection is +//! dropped. There is no framing negotiation — the first byte after the length +//! prefix is `"`. +//! +//! Observations are content-addressed, so IDs are exchanged as lowercase hex +//! (the hex form of the 16-byte `ObservationId`). Topics and payloads are +//! passed to `observe` the same way: as hex. +//! +//! # Sequence numbers +//! +//! `now` is the tines tick passed to `observe`/`belief`; the service reads it +//! from its own wall clock (`now_ms()`) and ignores any caller-supplied value, +//! so the wire request carries no timestamp. + +use std::fmt::Write as _; +use std::io::{Read, Write}; +use std::os::unix::net::UnixListener; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, Mutex}; + +use tines_core::{Node, Store, Transport}; + +const MAX_FRAME: usize = 1 << 20; // 1 MiB cap on a single JSON frame + +/// A confidence event handed out by the IPC `events` request. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +pub struct EventOut { + /// Hex of the `ObservationId` this confidence concerns. + pub id: String, + /// Hex of the `NodeId` of the peer that acknowledged. + pub peer: String, +} + +/// A belief returned by the IPC `belief` request. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +pub struct BeliefOut { + /// Hex of the `ObservationId`. + pub id: String, + /// Hex of the topic. + pub topic: String, + /// Hex of the payload. + pub payload: String, +} + +/// An IPC request. See the module docs for the wire format. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +#[serde(tag = "kind")] +pub enum Request { + /// Insert an observation under `topic`. + Observe { + /// Hex of the topic bytes to observe. + topic: String, + /// Hex of the observation payload. + payload: String, + }, + /// Read the latest belief for `topic`. + Belief { + /// Hex of the topic bytes to query. + topic: String, + }, + /// Drain outstanding confidence events. + Events, + /// Report the node's identity and log depth. + Status, +} + +/// An IPC response. See the module docs for the wire format. +#[derive(serde::Serialize, serde::Deserialize, Debug)] +#[serde(tag = "kind")] +pub enum Response { + /// The inserted observation's `id` (hex). + Observe { + /// Hex of the `ObservationId`. + id: String, + }, + /// The latest belief for the topic, if any. + Belief { + /// The stored belief, or `None` if the topic is unknown. + obs: Option, + }, + /// Confidence events drained from the exchange. + Events { + /// The drained confidence events. + events: Vec, + }, + /// The node's `id`, next tines `next_seq`, and observations `stored`. + Status { + /// Hex of the node's own `NodeId`. + id: String, + /// Next tines sequence number the node will use. + next_seq: u64, + /// Number of observations in the store. + stored: u32, + }, + /// The request was malformed or rejected. + Error { + /// A human-readable explanation. + message: String, + }, +} + +/// Handle one IPC request against a node. +/// +/// This is the single point where a request becomes a node operation; +/// [`serve`](fn@serve) decodes frames, calls `handle`, and re-encodes the +/// response. Kept generic over the store/transport so tests can drive it with a +/// mock transport. +pub fn handle(node: &mut Node, req: Request, now: u64) -> Response +where + S: Store + 'static, + T: Transport, +{ + match req { + Request::Observe { topic, payload } => { + let topic = match decode_hex(&topic) { + Ok(t) => t, + Err(e) => return error(e), + }; + let payload = match decode_hex(&payload) { + Ok(t) => t, + Err(e) => return error(e), + }; + match node.observe(&topic, &payload, None, now) { + Ok(id) => Response::Observe { + id: encode_hex(&id), + }, + Err(e) => error(format!("observe failed: {e:?}")), + } + } + Request::Belief { topic } => { + let topic = match decode_hex(&topic) { + Ok(t) => t, + Err(e) => return error(e), + }; + match node.belief(&topic, &tines_core::LatestWins, now) { + Some(obs) => Response::Belief { + obs: Some(BeliefOut { + id: encode_hex(&obs.id()), + topic: encode_hex(&obs.topic), + payload: encode_hex(&obs.payload), + }), + }, + None => Response::Belief { obs: None }, + } + } + Request::Events => { + let events = node + .drain_events() + .into_iter() + .map(|ev| EventOut { + id: encode_hex(&ev.id), + peer: encode_hex(&ev.peer), + }) + .collect(); + Response::Events { events } + } + Request::Status => Response::Status { + id: encode_hex(&node.id()), + next_seq: node.next_seq(), + stored: node.store().len() as u32, + }, + } +} + +/// Serve IPC on `socket_path` until the listener dies or a response write +/// fails. Blocks; each accepted connection is served request-by-request. +/// +/// `Node` may be any concrete store/transport, so the real binary passes +/// `Node` and tests pass a mock. +pub fn serve(node: Node, socket_path: impl AsRef) -> std::io::Result<()> +where + S: Store + 'static, + T: Transport + 'static, +{ + let node = Arc::new(Mutex::new(node)); + let path: PathBuf = socket_path.as_ref().to_path_buf(); + + // Clear a stale socket file left by a previous run so bind() can't fail. + let _ = std::fs::remove_file(&path); + let listener = UnixListener::bind(&path)?; + + for stream in listener.incoming() { + let mut stream = stream?; + // Serve this connection request-by-request until the peer disconnects + // or sends a malformed frame, then move on to the next connection. + while let Ok(req) = read_request(&mut stream) { + let now = now_ms(); + let resp = { + let mut guard = node.lock().unwrap(); + handle(&mut guard, req, now) + }; + // A client that vanished mid-connection shouldn't take down the + // whole service; just drop this one socket. + if write_response(&mut stream, &resp).is_err() { + break; + } + } + } + Ok(()) +} + +/// Read one length-prefixed JSON frame from a stream. +fn read_request(stream: &mut S) -> std::io::Result { + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf)?; + let len = u32::from_le_bytes(len_buf) as usize; + if len > MAX_FRAME { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + format!("frame {len} bytes exceeds {MAX_FRAME}"), + )); + } + let mut buf = vec![0u8; len]; + stream.read_exact(&mut buf)?; + serde_json::from_slice(&buf) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e)) +} + +/// Write one length-prefixed JSON frame to a stream. +fn write_response(stream: &mut S, resp: &Response) -> std::io::Result<()> { + let body = serde_json::to_vec(resp) + .map_err(|e| std::io::Error::new(std::io::ErrorKind::InvalidData, e))?; + if body.len() > MAX_FRAME { + return Err(std::io::Error::new( + std::io::ErrorKind::InvalidData, + "response too large", + )); + } + let len = (body.len() as u32).to_le_bytes(); + stream.write_all(&len)?; + stream.write_all(&body)?; + stream.flush() +} + +fn error(msg: impl std::fmt::Display) -> Response { + Response::Error { + message: msg.to_string(), + } +} + +fn now_ms() -> u64 { + std::time::SystemTime::now() + .duration_since(std::time::SystemTime::UNIX_EPOCH) + .map(|d| d.as_secs()) + .unwrap_or(0) +} + +fn encode_hex(bytes: &[u8]) -> String { + let mut s = String::with_capacity(bytes.len() * 2); + for b in bytes { + let _ = write!(s, "{b:02x}"); + } + s +} + +fn decode_hex(s: &str) -> Result, String> { + if !s.len().is_multiple_of(2) { + return Err(format!("odd-length hex string: {}", s.len())); + } + let bytes = s.as_bytes(); + let mut out = Vec::with_capacity(bytes.len() / 2); + let mut i = 0; + while i < bytes.len() { + let hi = hex_val(bytes[i])?; + let lo = hex_val(bytes[i + 1])?; + out.push((hi << 4) | lo); + i += 2; + } + Ok(out) +} + +fn hex_val(c: u8) -> Result { + match c { + b'0'..=b'9' => Ok(c - b'0'), + b'a'..=b'f' => Ok(c - b'a' + 10), + b'A'..=b'F' => Ok(c - b'A' + 10), + other => Err(format!("invalid hex char: {other:02x}")), + } +} diff --git a/tines/src/lib.rs b/tines/src/lib.rs new file mode 100644 index 00000000..607196ce --- /dev/null +++ b/tines/src/lib.rs @@ -0,0 +1,15 @@ +//! The standalone `tines` service. +//! +//! Hosts a `Node` over a Unix-domain socket and serves a +//! small request/response IPC for `observe` / `belief` / `events` / `status`. +//! One instance runs per device — the binary (`src/main.rs`) wires a +//! `Node` for real BLE, while tests drive the same +//! [`serve`] / [`handle`] path with a mock transport so no radio +//! is required. +//! +//! The IPC protocol is documented in [`ipc`]. +#![deny(missing_docs)] + +pub mod ipc; + +pub use ipc::{handle, serve, BeliefOut, EventOut, Request, Response}; diff --git a/tines/src/main.rs b/tines/src/main.rs new file mode 100644 index 00000000..8cb47948 --- /dev/null +++ b/tines/src/main.rs @@ -0,0 +1,88 @@ +//! Standalone `tines` service binary. +//! +//! Loads a JSON config, builds a `Node` (BLE over +//! `bluer`), and serves the IPC ([`tines::serve`]) over a Unix-domain socket. +//! One instance per device — the typical deployment on robots without ROS2. +//! +//! The node/peer ids are 32 lowercase hex chars (the 16-byte `NodeId`), and +//! `first_seq` is the next tines sequence number to use (persisted across +//! reboots, starting from `tines`'s 1-based numbering). + +use std::error::Error; +use std::path::Path; + +use tines::serve; +use tines_bt::{BluerRadio, BtTransport}; +use tines_core::{Node, NodeId}; +use tines_store::LogStore; + +/// On-disk configuration for a running service. +#[derive(serde::Deserialize)] +struct Config { + /// This device's advertised `NodeId`, 32 lowercase hex chars. + node_id: String, + /// The peer's advertised `NodeId`, 32 lowercase hex chars. + peer: String, + /// Next tines sequence number to use (persisted across reboots). + first_seq: u64, + /// Absolute path to the LogStore append-only log. + log_path: String, + /// Unix-domain socket path the IPC listens on. + socket_path: String, +} + +fn main() -> Result<(), Box> { + let cfg_path = std::env::args() + .nth(1) + .ok_or("usage: tines ")?; + let config: Config = serde_json::from_str(&std::fs::read_to_string(cfg_path)?)?; + + let me = parse_node_id(&config.node_id)?; + let peer = parse_node_id(&config.peer)?; + + // LogStore is the Store (S); BtTransport wraps it as the Transport (T). + let store = LogStore::new(Path::new(&config.log_path), 0, config.first_seq) + .map_err(|e| format!("failed to open log store: {e:?}"))?; + let radio = BluerRadio::new(me, peer); + let transport = BtTransport::new(me, peer, radio); + let node = Node::new(me, config.first_seq, store, transport); + + println!( + "tines: listening on {} (node {})", + config.socket_path, + node_hex(me) + ); + serve(node, Path::new(&config.socket_path))?; + Ok(()) +} + +fn parse_node_id(s: &str) -> Result> { + if s.len() != 32 { + return Err(format!("node id must be 32 hex chars, got {}", s.len()).into()); + } + let mut id = [0u8; 16]; + for (i, chunk) in s.as_bytes().chunks(2).enumerate() { + let hi = hex_val(chunk[0])?; + let lo = hex_val(chunk[1])?; + id[i] = (hi << 4) | lo; + } + Ok(id) +} + +fn hex_val(c: u8) -> Result> { + match c { + b'0'..=b'9' => Ok(c - b'0'), + b'a'..=b'f' => Ok(c - b'a' + 10), + b'A'..=b'F' => Ok(c - b'A' + 10), + other => Err(format!("invalid hex char: {other:02x}").into()), + } +} + +fn node_hex(id: NodeId) -> String { + use std::fmt::Write as _; + let mut s = String::with_capacity(32); + for b in id { + let _ = write!(s, "{b:02x}"); + } + s +} diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs new file mode 100644 index 00000000..b53e51a7 --- /dev/null +++ b/tines/tests/ipc.rs @@ -0,0 +1,315 @@ +//! Integration tests for the `tines` IPC. +//! +//! These drive `handle`/`serve` with a mock transport, so no real BLE radio is +//! needed: the mock surfaces a single `Contact`, records outbound frames, and +//! auto-acks any observation broadcast to it (simulating a well-behaved peer). + +use std::collections::{HashSet, VecDeque}; +use std::io::{Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process; +use std::thread; +use std::time::Duration; + +use heapless::Vec as HeaplessVec; +use std::fmt::Write as _; +use tines::{handle, serve, Request, Response}; +use tines_core::{ + consts, Event, Frame, FrameBuf, Node, NodeId, ObservationId, SendError, Transport, +}; +use tines_store::LogStore; + +/// A no-op [`Transport`] for tests: it never surfaces inbound frames except +/// the ones injected by the test, and it records outbound frames. +struct MockTransport { + peer: NodeId, + contacted: bool, + pending: VecDeque>, + sent: Vec>, + acked: HashSet, +} + +/// A well-behaved peer: auto-acks each distinct observation it is sent. +impl MockTransport { + fn new(peer: NodeId) -> Self { + Self { + peer, + contacted: false, + pending: VecDeque::new(), + sent: Vec::new(), + acked: HashSet::new(), + } + } +} + +impl Transport for MockTransport { + type Addr = NodeId; + + fn poll(&mut self) -> Option> { + if !self.contacted { + self.contacted = true; + return Some(Event::Contact(self.peer)); + } + self.pending.pop_front() + } + + fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> { + self.sent.push(frame.to_vec()); + match Frame::decode(frame) { + Ok(Frame::Obs { obs, .. }) => { + // A well-behaved peer acks each distinct observed broadcast. + if self.acked.insert(obs.id()) { + let ack = ack_frame(*to, obs.id()); + self.pending.push_back(frame_event(*to, &ack)); + } + } + Ok(Frame::Summary { salt, .. }) => { + // Echo back an empty, full-size Summary so the exchange learns + // the peer holds nothing and broadcasts its diff (which we then + // ack, firing a ConfidenceEvent). + let raw: [u8; consts::SUMMARY_BYTES] = [0u8; consts::SUMMARY_BYTES]; + let bits: HeaplessVec = + HeaplessVec::from_slice(&raw).expect("512 fits SUMMARY_BYTES"); + let reply = Frame::Summary { + from: *to, + salt, + bits, + }; + self.pending.push_back(frame_event(*to, &reply)); + } + _ => {} + } + Ok(()) + } + + fn mtu(&self) -> usize { + consts::MAX_FRAME + } +} + +fn ack_frame(from: NodeId, id: ObservationId) -> Frame { + let ids: HeaplessVec = + HeaplessVec::from_slice(&[id]).expect("one id fits ACK_IDS_CAP"); + Frame::Ack { from, ids } +} + +fn frame_event(from: NodeId, frame: &Frame) -> Event { + let mut buf = [0u8; consts::MAX_FRAME]; + let n = frame.encode(&mut buf).expect("frame fits MAX_FRAME"); + Event::Frame( + from, + FrameBuf::from_slice(&buf[..n]).expect("frame fits FrameBuf"), + ) +} + +fn topic() -> Vec { + b"tines/test".to_vec() +} + +fn payload() -> Vec { + b"hello".to_vec() +} + +fn hex_encode(bytes: &[u8]) -> String { + let mut s = String::with_capacity(bytes.len() * 2); + for b in bytes { + let _ = write!(s, "{b:02x}"); + } + s +} + +fn unique_dir(tag: &str) -> PathBuf { + let dir = std::env::temp_dir().join(format!("tines-ipc-{}-{}", tag, process::id())); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +fn build_node(peer: NodeId, dir: &Path) -> Node { + let store_path = dir.join("store.log"); + let store = LogStore::new(&store_path, 0, 1).expect("store open"); + Node::new(peer, 1, store, MockTransport::new(peer)) +} + +fn connect(path: &Path) -> UnixStream { + let mut delay = 5; + for _ in 0..500 { + if let Ok(s) = UnixStream::connect(path) { + return s; + } + thread::sleep(Duration::from_millis(delay)); + delay = delay.min(50); + } + panic!("failed to connect to {}", path.display()); +} + +fn rpc(stream: &mut S, req: &Request) { + let body = serde_json::to_vec(req).expect("serialize request"); + let len = (body.len() as u32).to_le_bytes(); + stream.write_all(&len).unwrap(); + stream.write_all(&body).unwrap(); + stream.flush().unwrap(); +} + +fn read_response(stream: &mut S) -> Response { + let mut len_buf = [0u8; 4]; + stream.read_exact(&mut len_buf).unwrap(); + let len = u32::from_le_bytes(len_buf) as usize; + let mut buf = vec![0u8; len]; + stream.read_exact(&mut buf).unwrap(); + serde_json::from_slice(&buf).expect("valid response") +} + +#[test] +fn handle_status_and_belief_direct() { + let dir = unique_dir("direct"); + let mut node = build_node([3; 16], &dir); + let x = node.observe(&topic(), &payload(), None, 0).unwrap(); + + let status = handle(&mut node, Request::Status, 0); + match status { + Response::Status { + id, + next_seq, + stored, + } => { + assert_eq!(id, hex_encode(&node.id())); + assert_eq!(next_seq, 2); + assert_eq!(stored, 1); + } + other => panic!("expected status, got {other:?}"), + } + + let belief = handle( + &mut node, + Request::Belief { + topic: hex_encode(&topic()), + }, + 0, + ); + match belief { + Response::Belief { obs: Some(b) } => { + assert_eq!(b.id, hex_encode(&x)); + assert_eq!(b.topic, hex_encode(&topic())); + assert_eq!(b.payload, hex_encode(&payload())); + } + other => panic!("expected belief, got {other:?}"), + } +} + +#[test] +fn handle_unknown_obs_is_belief_none() { + let dir = unique_dir("direct"); + let mut node = build_node([3; 16], &dir); + let resp = handle( + &mut node, + Request::Belief { + topic: hex_encode(&topic()), + }, + 0, + ); + match resp { + Response::Belief { obs: None } => {} + other => panic!("expected empty belief, got {other:?}"), + } +} + +#[test] +fn observe_and_belief_over_socket() { + let dir = unique_dir("roundtrip"); + let socket = dir.join("ipc.sock"); + let node = build_node([9; 16], &dir); + let serve_path = socket.clone(); + let _handle = thread::spawn(move || serve(node, serve_path).unwrap()); + + let mut stream = connect(&socket); + + rpc( + &mut stream, + &Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(&payload()), + }, + ); + let id = match read_response(&mut stream) { + Response::Observe { id } => id, + other => panic!("expected Observe, got {other:?}"), + }; + assert_eq!(id.len(), 32); + + rpc( + &mut stream, + &Request::Belief { + topic: hex_encode(&topic()), + }, + ); + match read_response(&mut stream) { + Response::Belief { obs: Some(b) } => { + assert_eq!(b.id, id); + assert_eq!(b.topic, hex_encode(&topic())); + assert_eq!(b.payload, hex_encode(&payload())); + } + other => panic!("expected Belief, got {other:?}"), + } +} + +#[test] +fn events_over_socket() { + let dir = unique_dir("events"); + let socket = dir.join("ipc.sock"); + let peer = [7; 16]; + let mut node = build_node(peer, &dir); + + // Drive a confidence event into the node directly: observe (broadcast), + // let the mock auto-ack it, and poll it out of the exchange. + let x = node.observe(&topic(), &payload(), None, 0).unwrap(); + node.poll(0); // contact -> node emits its summary + node.poll(0); // node broadcasts the observed topic; mock injects the ack + node.poll(0); // ack processed -> confidence event fires + + let serve_path = socket.clone(); + let _handle = thread::spawn(move || serve(node, serve_path).unwrap()); + let mut stream = connect(&socket); + + rpc(&mut stream, &Request::Events); + match read_response(&mut stream) { + Response::Events { events } => { + assert_eq!(events.len(), 1, "expected one confidence event"); + assert_eq!(events[0].id, hex_encode(&x), "event id matches observed id"); + assert_eq!(events[0].peer, hex_encode(&peer), "event came from peer"); + } + other => panic!("expected Events, got {other:?}"), + } +} + +#[test] +fn handle_bad_hex_is_error() { + let dir = unique_dir("hex"); + let mut node = build_node([3; 16], &dir); + let resp = handle( + &mut node, + Request::Observe { + topic: "z".into(), + payload: "ff".into(), + }, + 0, + ); + match resp { + Response::Error { .. } => {} + other => panic!("expected error, got {other:?}"), + } +} + +#[test] +fn confidence_event_direct_probe() { + let dir = unique_dir("ev"); + let mut node = build_node([7; 16], &dir); + let x = node.observe(&topic(), &payload(), None, 0).unwrap(); + node.poll(0); + node.poll(0); + node.poll(0); + let evs = node.drain_events(); + assert_eq!(evs.len(), 1, "one confidence event after observe+ack"); + assert_eq!(evs[0].id, x); + assert_eq!(evs[0].peer, [7; 16]); +}