diff --git a/Cargo.lock b/Cargo.lock index 9ce9397d..56419165 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1469,6 +1469,7 @@ dependencies = [ "serde", "serde_json", "tines-bt", + "tines-bus", "tines-core", "tines-store", ] diff --git a/tines-bus/src/daemon.rs b/tines-bus/src/daemon.rs index 611d1cab..bb51c7b0 100644 --- a/tines-bus/src/daemon.rs +++ b/tines-bus/src/daemon.rs @@ -221,6 +221,15 @@ fn process_msg(idx: usize, payload: &[u8], conns: &mut [Conn], actions: &mut Vec }); } } + // ...and the newcomer learns everyone already here. + for c in before.iter().chain(after.iter()) { + if let Some(existing) = c.id { + actions.push(Action::Forward { + to: idx, + env: control(0, existing), + }); + } + } } } Some(my_id) => { diff --git a/tines-bus/tests/daemon.rs b/tines-bus/tests/daemon.rs index 49cc2764..ae82063c 100644 --- a/tines-bus/tests/daemon.rs +++ b/tines-bus/tests/daemon.rs @@ -78,6 +78,21 @@ fn expect_no_data(s: &mut UnixStream, window_ms: u64) { } } +#[test] +fn newcomer_learns_existing_peers() { + let dir = unique_dir("newcomer"); + let sock = spawn_daemon(&dir); + + let mut a = connect(&sock); + hello(&mut a, &[1; 16]); + + // B joins after A: B must be told A is already there. + let mut b = connect(&sock); + hello(&mut b, &[2; 16]); + expect_control(&mut a, 0, &[2; 16]); + expect_control(&mut b, 0, &[1; 16]); +} + #[test] fn duplicate_hello_replaces() { let dir = unique_dir("dup"); @@ -113,6 +128,9 @@ fn mismatched_from_is_dropped() { let mut b = connect(&sock); hello(&mut b, &[2; 16]); + // B learns about A on join (the newcomer announcement). + expect_control(&mut b, 0, &[1; 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); @@ -142,6 +160,8 @@ fn unknown_dest_is_dropped() { hello(&mut b, &[2; 16]); // Wait until the daemon has registered B: A sees the join control. expect_control(&mut a, 0, &[2; 16]); + // B's side of the same join: the newcomer learns A is already here. + expect_control(&mut b, 0, &[1; 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); @@ -158,8 +178,10 @@ fn oversized_frame_closes_connection() { let mut b = connect(&sock); hello(&mut b, &[2; 16]); - // Handshake: once A has seen B join, both hellos are registered. + // Handshake: once A has seen B join, both hellos are registered, and + // B has learned A on its side too. expect_control(&mut a, 0, &[2; 16]); + expect_control(&mut b, 0, &[1; 16]); // A frame one byte over MAX_FRAME: that connection is closed... data(&mut a, &[1; 16], &[2; 16], &vec![0u8; MAX_FRAME + 1]); diff --git a/tines-bus/tests/transport.rs b/tines-bus/tests/transport.rs index b5203eb1..7e3d8e7b 100644 --- a/tines-bus/tests/transport.rs +++ b/tines-bus/tests/transport.rs @@ -102,6 +102,16 @@ fn spoofed_frame_is_dropped() { Duration::from_secs(3), ); + // B's side of the same join: the newcomer learns A is there. + match poll_until( + &mut b, + |ev| matches!(ev, Event::Contact(_)), + Duration::from_secs(3), + ) { + Event::Contact(id) => assert_eq!(id, [1; 16]), + other => panic!("expected Contact, got {other:?}"), + } + // 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]); diff --git a/tines/Cargo.toml b/tines/Cargo.toml index 53df5afc..3faf0696 100644 --- a/tines/Cargo.toml +++ b/tines/Cargo.toml @@ -14,6 +14,7 @@ categories = ["science::robotics", "network-programming"] tines-core = { path = "../tines-core", version = "1" } tines-store = { path = "../tines-store", version = "0.4.0" } tines-bt = { path = "../tines-bt", version = "0.4.0" } +tines-bus = { path = "../tines-bus", version = "0.4.0" } serde = { version = "1", features = ["derive"] } serde_json = "1" diff --git a/tines/src/config.rs b/tines/src/config.rs new file mode 100644 index 00000000..afac7de3 --- /dev/null +++ b/tines/src/config.rs @@ -0,0 +1,194 @@ +//! On-disk service configuration, shared by the `tines` binary and its +//! clients (the fleet console generates these configs and drives the service +//! it configures). +//! +//! The config is JSON with `deny_unknown_fields`; every key except the core +//! `node_id` / `log_path` / `socket_path` is optional. See [`Config`] for the +//! per-key contract, and [`Config::validate`] for the transport constraints. + +use crate::trust::DEFAULT_PEER_TIME_WINDOW_MS; + +/// On-disk configuration for a running service. Unknown keys are rejected. +#[derive(Debug, serde::Deserialize)] +#[serde(deny_unknown_fields)] +pub struct Config { + /// This device's advertised `NodeId`, 32 lowercase hex chars. + pub node_id: String, + /// The peer's advertised `NodeId`, 32 lowercase hex chars. Required for + /// the bt transport (its single peer); for the bus transport it is the + /// allowlist of one — `None` means any bus peer. + pub peer: Option, + /// Absolute path to the LogStore append-only log. + pub log_path: String, + /// Unix-domain socket path the IPC listens on. + pub socket_path: String, + /// Exchange poll interval in milliseconds. The real default belongs to the + /// builder; 20 ms keeps the link responsive without burning idle cycles. + #[serde(default = "default_poll_interval_ms")] + pub poll_interval_ms: u64, + /// Trust the system clock for the host's wall time, instead of deriving it + /// from a peer's observation. Set to `"system"` when the machine's clock is + /// reliable — an RTC, or a GPS/NTP-disciplined clock — the operator asserts + /// it rather than tines detecting it. Left unset, tines falls back to a + /// peer observation, and, with none available, stays quarantined. Any + /// other value is a config error. + #[serde(default)] + pub trusted_clock: Option, + /// How long, from the first plausible peer estimate, to keep collecting + /// estimates before adopting the latest (spec decision J). + #[serde(default = "default_peer_time_window_ms")] + pub peer_time_window_ms: u64, + /// The `unknown_downtime_age_ms` broadcast floor passed to LogStore while + /// the wall clock is unknown (the fold's unknown-downtime branch). Optional; + /// leave the LogStore default (`24h`) when unset. + /// + /// The floor must stay below the host's trim `max_age_ms`. That bound waits + /// for host hardening — the trim defaults are still empty — so this check is + /// not validated until then. + #[serde(default)] + pub unknown_downtime_age_ms: Option, + /// Which transport the service runs: `bt` (default) or `bus`. + #[serde(default)] + pub transport: Option, + /// Absolute path to the bus daemon's Unix-domain socket. Required when + /// `transport` is `bus`. + #[serde(default)] + pub bus_socket: Option, +} + +impl Config { + /// Check the transport constraints. Called by `main` before anything + /// else touches the config. + /// + /// - `bt` (the default) requires `peer`. + /// - `bus` requires `bus_socket`; `peer` is the allowlist of one, and + /// `None` means any peer. + pub fn validate(&self) -> Result<(), String> { + match self.transport.unwrap_or(TransportKind::Bt) { + TransportKind::Bt if self.peer.is_none() => Err("transport bt requires peer".into()), + TransportKind::Bus if self.bus_socket.is_none() => { + Err("transport bus requires bus_socket".into()) + } + _ => Ok(()), + } + } +} + +/// The accepted values of `trusted_clock`. +#[derive(Debug, PartialEq, Eq, serde::Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum TrustedClock { + /// Trust the system clock (`"system"`). + System, +} + +/// The transport the service runs over. +#[derive(Debug, PartialEq, Eq, Clone, Copy, serde::Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum TransportKind { + /// The BLE transport (the default), with the single configured peer. + Bt, + /// The host bus transport, peers = every node on the bus daemon (or the + /// allowlisted one). + Bus, +} + +fn default_poll_interval_ms() -> u64 { + 20 +} + +fn default_peer_time_window_ms() -> u64 { + DEFAULT_PEER_TIME_WINDOW_MS +} + +#[cfg(test)] +mod tests { + use super::{Config, TransportKind, TrustedClock}; + + const BASE: &str = r#""node_id": "00000000000000000000000000000001", + "peer": "00000000000000000000000000000002", + "log_path": "/var/lib/tines/log", + "socket_path": "/run/tines.sock""#; + + fn parse(extra: &str) -> Result { + serde_json::from_str(&format!("{{{BASE}{extra}}}")) + } + + #[test] + fn defaults() { + let c = parse("").expect("minimal config"); + assert_eq!(c.poll_interval_ms, 20); + assert_eq!(c.peer_time_window_ms, 10_000); + assert_eq!(c.trusted_clock, None); + assert_eq!(c.unknown_downtime_age_ms, None); + } + + #[test] + fn trusted_clock_system_is_accepted() { + let c = parse(r#", "trusted_clock": "system""#).expect("system"); + assert_eq!(c.trusted_clock, Some(TrustedClock::System)); + } + + #[test] + fn trusted_clock_unknown_value_is_rejected() { + assert!(parse(r#", "trusted_clock": "System""#).is_err()); + assert!(parse(r#", "trusted_clock": "rtc""#).is_err()); + assert!(parse(r#", "trusted_clock": "systme""#).is_err()); + } + + #[test] + fn unknown_field_is_rejected() { + assert!(parse(r#", "trusted_clok": "system""#).is_err()); + } + + #[test] + fn peer_time_window_is_configurable() { + let c = parse(r#", "peer_time_window_ms": 2500"#).expect("window"); + assert_eq!(c.peer_time_window_ms, 2_500); + } + + const BUS_BASE: &str = r#""node_id": "00000000000000000000000000000001", + "log_path": "/var/lib/tines/log", + "socket_path": "/run/tines.sock""#; + + fn parse_bus(extra: &str) -> Result { + serde_json::from_str(&format!("{{{BUS_BASE}{extra}}}")) + } + + #[test] + fn bus_config_parses() { + let c = parse_bus(r#", "transport": "bus", "bus_socket": "/run/tines-bus.sock""#) + .expect("bus config"); + assert_eq!(c.transport, Some(TransportKind::Bus)); + assert_eq!(c.bus_socket.as_deref(), Some("/run/tines-bus.sock")); + assert_eq!(c.peer, None); + assert_eq!(c.validate(), Ok(())); + } + + #[test] + fn bt_without_peer_is_invalid() { + let c = parse_bus("").expect("bt config without peer parses"); + assert_eq!(c.validate(), Err("transport bt requires peer".to_string())); + } + + #[test] + fn bus_without_socket_is_invalid() { + let c = parse_bus(r#", "transport": "bus""#).expect("parses"); + assert_eq!( + c.validate(), + Err("transport bus requires bus_socket".to_string()) + ); + } + + #[test] + fn unknown_transport_is_rejected() { + assert!(parse_bus(r#", "transport": "wifi""#).is_err()); + } + + #[test] + fn transport_defaults_to_bt() { + let c = parse("").expect("minimal config"); + assert_eq!(c.transport, None); + assert_eq!(c.validate(), Ok(())); + } +} diff --git a/tines/src/lib.rs b/tines/src/lib.rs index ba3090cc..99bf943f 100644 --- a/tines/src/lib.rs +++ b/tines/src/lib.rs @@ -3,19 +3,24 @@ //! 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. +//! `Node` for real BLE or the host bus, while tests +//! drive the same [`serve`] / [`handle`] path with a mock transport so no +//! radio is required. //! -//! The IPC protocol is documented in [`ipc`]. +//! The IPC protocol is documented in [`ipc`]; the on-disk configuration in +//! [`config`]. #![deny(missing_docs)] pub mod clock; +pub mod config; pub mod ipc; +pub mod transport; pub mod trust; pub use clock::HostClock; +pub use config::{Config, TransportKind, TrustedClock}; pub use ipc::{ handle, serve, BeliefOut, ContactOut, DropOut, EventOut, ObservationOut, Request, Response, LIST_CAP, }; +pub use transport::HostTransport; diff --git a/tines/src/main.rs b/tines/src/main.rs index c17e5332..692024e6 100644 --- a/tines/src/main.rs +++ b/tines/src/main.rs @@ -1,7 +1,8 @@ //! 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. +//! Loads a JSON config, builds a `Node` (BLE over +//! `bluer`, or the host bus — see [`tines::config::TransportKind`]), and +//! serves the IPC ([`tines::serve`]) over a Unix-domain socket. //! One instance per device — the typical deployment on robots without ROS2. //! //! The node and peer ids are 32 lowercase hex chars (the 16-byte `NodeId`). @@ -11,14 +12,16 @@ //! # Config //! //! A JSON object; an unknown key is an error, so a typo can't silently fall -//! back to a default. +//! back to a default. The full contract lives in [`tines::config::Config`]. //! //! | key | required | meaning | //! |---|---|---| //! | `node_id` | yes | this device's `NodeId`, 32 hex chars | -//! | `peer` | yes | the peer's `NodeId`, 32 hex chars | +//! | `peer` | bt only | the peer's `NodeId`, 32 hex chars; on bus, the allowlist of one (`None` = any bus peer) | //! | `log_path` | yes | the LogStore log file | //! | `socket_path` | yes | the IPC Unix socket | +//! | `transport` | no (`bt`) | `"bt"` or `"bus"` | +//! | `bus_socket` | bus only | the bus daemon's Unix socket | //! | `poll_interval_ms` | no (20) | exchange poll cadence | //! | `trusted_clock` | no (unset) | `"system"` to trust the system clock; the only accepted value | //! | `peer_time_window_ms` | no (10000) | how long to collect peer time estimates before adopting the latest | @@ -50,81 +53,32 @@ use std::error::Error; use std::path::Path; -use tines::trust::{adopt_wall_clock, system_wall_ms, DEFAULT_PEER_TIME_WINDOW_MS}; +use tines::config::{Config, TransportKind, TrustedClock}; +use tines::transport::HostTransport; +use tines::trust::{adopt_wall_clock, system_wall_ms}; use tines::{serve, HostClock}; use tines_bt::{BluerRadio, BtTransport}; +use tines_bus::BusTransport; use tines_core::{Node, NodeId}; use tines_store::LogStore; -/// On-disk configuration for a running service. Unknown keys are rejected. -#[derive(Debug, serde::Deserialize)] -#[serde(deny_unknown_fields)] -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, - /// Absolute path to the LogStore append-only log. - log_path: String, - /// Unix-domain socket path the IPC listens on. - socket_path: String, - /// Exchange poll interval in milliseconds. The real default belongs to the - /// builder; 20 ms keeps the link responsive without burning idle cycles. - #[serde(default = "default_poll_interval_ms")] - poll_interval_ms: u64, - /// Trust the system clock for the host's wall time, instead of deriving it - /// from a peer's observation. Set to `"system"` when the machine's clock is - /// reliable — an RTC, or a GPS/NTP-disciplined clock — the operator asserts - /// it rather than tines detecting it. Left unset, tines falls back to a - /// peer observation, and, with none available, stays quarantined. Any - /// other value is a config error. - #[serde(default)] - trusted_clock: Option, - /// How long, from the first plausible peer estimate, to keep collecting - /// estimates before adopting the latest (spec decision J). - #[serde(default = "default_peer_time_window_ms")] - peer_time_window_ms: u64, - /// The `unknown_downtime_age_ms` broadcast floor passed to LogStore while - /// the wall clock is unknown (the fold's unknown-downtime branch). Optional; - /// leave the LogStore default (`24h`) when unset. - /// - /// The floor must stay below the host's trim `max_age_ms`. That bound waits - /// for host hardening — the trim defaults are still empty — so this check is - /// not validated until then. - #[serde(default)] - unknown_downtime_age_ms: Option, -} - -/// The accepted values of `trusted_clock`. -#[derive(Debug, PartialEq, Eq, serde::Deserialize)] -#[serde(rename_all = "lowercase")] -enum TrustedClock { - /// Trust the system clock (`"system"`). - System, -} - -fn default_poll_interval_ms() -> u64 { - 20 -} - -fn default_peer_time_window_ms() -> u64 { - DEFAULT_PEER_TIME_WINDOW_MS -} - 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)?) .map_err(|e| format!("invalid config: {e}"))?; + config + .validate() + .map_err(|e| format!("invalid config: {e}"))?; 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). - // unknown_downtime_age_ms (the broadcast floor while the wall clock is - // unknown) is optional; leave the LogStore default when unset. - // Peer estimates outside the plausible range don't count (decision J). + // LogStore is the Store (S); HostTransport wraps BLE or the bus as the + // Transport (T). unknown_downtime_age_ms (the broadcast floor while the + // wall clock is unknown) is optional; leave the LogStore default when + // unset. Peer estimates outside the plausible range don't count + // (decision J). let builder = LogStore::builder(Path::new(&config.log_path)) .plausible_wall_ms(tines::trust::PLAUSIBLE_WALL_MS); let builder = match config.unknown_downtime_age_ms { @@ -136,8 +90,26 @@ fn main() -> Result<(), Box> { .open(clock.now()) .map_err(|e| format!("failed to open log store: {e:?}"))?; - let radio = BluerRadio::new(me, peer); - let transport = BtTransport::new(me, peer, radio); + // bt: the single configured peer (validation guarantees it). bus: the + // configured peer is the allowlist of one; None means any bus peer. + let transport = match config.transport.unwrap_or(TransportKind::Bt) { + TransportKind::Bt => { + let peer = parse_node_id(config.peer.as_deref().expect("validated"))?; + let radio = BluerRadio::new(me, peer); + HostTransport::Bt(BtTransport::new(me, peer, radio)) + } + TransportKind::Bus => { + let allowlist = match &config.peer { + Some(p) => Some(parse_node_id(p)?), + None => None, + }; + HostTransport::Bus(BusTransport::new( + me, + allowlist, + config.bus_socket.as_deref().expect("validated"), + )?) + } + }; // `serve` runs the exchange inline (a poll loop fed by the socket read // timeouts) on this same thread, so observations and beliefs flow over the // transport, not just over IPC. The node isn't `Send` (the BLE transport @@ -212,50 +184,3 @@ fn node_hex(id: NodeId) -> String { } s } - -#[cfg(test)] -mod tests { - use super::{Config, TrustedClock}; - - const BASE: &str = r#""node_id": "00000000000000000000000000000001", - "peer": "00000000000000000000000000000002", - "log_path": "/var/lib/tines/log", - "socket_path": "/run/tines.sock""#; - - fn parse(extra: &str) -> Result { - serde_json::from_str(&format!("{{{BASE}{extra}}}")) - } - - #[test] - fn defaults() { - let c = parse("").expect("minimal config"); - assert_eq!(c.poll_interval_ms, 20); - assert_eq!(c.peer_time_window_ms, 10_000); - assert_eq!(c.trusted_clock, None); - assert_eq!(c.unknown_downtime_age_ms, None); - } - - #[test] - fn trusted_clock_system_is_accepted() { - let c = parse(r#", "trusted_clock": "system""#).expect("system"); - assert_eq!(c.trusted_clock, Some(TrustedClock::System)); - } - - #[test] - fn trusted_clock_unknown_value_is_rejected() { - assert!(parse(r#", "trusted_clock": "System""#).is_err()); - assert!(parse(r#", "trusted_clock": "rtc""#).is_err()); - assert!(parse(r#", "trusted_clock": "systme""#).is_err()); - } - - #[test] - fn unknown_field_is_rejected() { - assert!(parse(r#", "trusted_clok": "system""#).is_err()); - } - - #[test] - fn peer_time_window_is_configurable() { - let c = parse(r#", "peer_time_window_ms": 2500"#).expect("window"); - assert_eq!(c.peer_time_window_ms, 2_500); - } -} diff --git a/tines/src/transport.rs b/tines/src/transport.rs new file mode 100644 index 00000000..9a842e83 --- /dev/null +++ b/tines/src/transport.rs @@ -0,0 +1,39 @@ +//! The host's chosen transport: either the BLE radio ([`BtTransport`]) or the +//! host bus ([`BusTransport`]), per the service config. + +use tines_bt::BtTransport; +use tines_bus::BusTransport; +use tines_core::{Event, NodeId, SendError, Transport}; + +/// The transport the service runs, selected by [`crate::config::Config`]. +pub enum HostTransport { + /// The BLE transport with the single configured peer. + Bt(BtTransport), + /// The host bus transport with the (optionally allowlisted) bus peers. + Bus(BusTransport), +} + +impl Transport for HostTransport { + type Addr = NodeId; + + fn poll(&mut self) -> Option> { + match self { + HostTransport::Bt(t) => t.poll(), + HostTransport::Bus(t) => t.poll(), + } + } + + fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> { + match self { + HostTransport::Bt(t) => t.send(to, frame), + HostTransport::Bus(t) => t.send(to, frame), + } + } + + fn mtu(&self) -> usize { + match self { + HostTransport::Bt(t) => t.mtu(), + HostTransport::Bus(t) => t.mtu(), + } + } +} diff --git a/tines/tests/service_bus.rs b/tines/tests/service_bus.rs new file mode 100644 index 00000000..b414ea43 --- /dev/null +++ b/tines/tests/service_bus.rs @@ -0,0 +1,215 @@ +//! End-to-end bus fleet test: two real `tines` service processes converge +//! over a live `tines-busd` — the sim-fleet deliverable without docker. +//! +//! Node A observes a broadcast; node B learns it over the bus (belief, store +//! list) and both see the contact change. + +use std::io::{Read, Write}; +use std::os::unix::net::UnixStream; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::sync::atomic::{AtomicUsize, Ordering}; +use std::time::{Duration, Instant}; + +use tines::{Request, Response}; + +static COUNTER: AtomicUsize = AtomicUsize::new(0); + +fn unique_dir(tag: &str) -> PathBuf { + let n = COUNTER.fetch_add(1, Ordering::Relaxed); + let dir = std::env::temp_dir().join(format!( + "tines-service-bus-{tag}-{}-{n}", + std::process::id() + )); + std::fs::create_dir_all(&dir).unwrap(); + dir +} + +fn node_id_hex(n: u8) -> String { + format!("{n:02x}").repeat(16) +} + +fn spawn_tines(config_path: &Path, dir: &Path) -> Child { + let stderr = std::fs::File::create(dir.join("stderr.log")).unwrap(); + Command::new(env!("CARGO_BIN_EXE_tines")) + .arg(config_path) + .stderr(Stdio::from(stderr)) + .stdout(Stdio::null()) + .spawn() + .expect("spawn tines service") +} + +fn connect(path: &Path, deadline: Instant) -> UnixStream { + loop { + if let Ok(s) = UnixStream::connect(path) { + return s; + } + assert!( + Instant::now() < deadline, + "failed to connect to {}", + path.display() + ); + std::thread::sleep(Duration::from_millis(20)); + } +} + +fn rpc(stream: &mut UnixStream, 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 UnixStream) -> 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") +} + +/// Poll `poll_once` until `pred` accepts a response, or fail after `deadline`. +fn poll_until( + mut poll_once: impl FnMut() -> Response, + pred: impl Fn(&Response) -> bool, + deadline: Instant, +) -> Response { + loop { + let r = poll_once(); + if pred(&r) { + return r; + } + assert!(Instant::now() < deadline, "timed out polling the service"); + std::thread::sleep(Duration::from_millis(50)); + } +} + +#[test] +fn two_services_converge_over_bus() { + let dir = unique_dir("converge"); + let bus_socket = dir.join("bus.sock"); + + // The bus daemon runs in-process on a thread (a thread in the test + // process, not an external binary, so the test has no CARGO_BIN_EXE + // dependency); poll until it has bound its socket. + let daemon_socket = bus_socket.clone(); + std::thread::spawn(move || { + let _ = tines_bus::daemon::serve(&daemon_socket); + }); + { + let mut bound = UnixStream::connect(&bus_socket); + let start = Instant::now(); + while bound.is_err() && start.elapsed() < Duration::from_secs(5) { + std::thread::sleep(Duration::from_millis(20)); + bound = UnixStream::connect(&bus_socket); + } + assert!(bound.is_ok(), "bus daemon did not bind its socket"); + } + + // Write both configs. + let mk_config = |node: &str, log: &str, sock: &str| -> String { + serde_json::json!({ + "node_id": node, + "log_path": dir.join(log).to_str().unwrap(), + "socket_path": dir.join(sock).to_str().unwrap(), + "trusted_clock": "system", + "transport": "bus", + "bus_socket": bus_socket.to_str().unwrap(), + }) + .to_string() + }; + let cfg_a = dir.join("a.json"); + let cfg_b = dir.join("b.json"); + std::fs::write(&cfg_a, mk_config(&node_id_hex(1), "a.log", "a.sock")).unwrap(); + std::fs::write(&cfg_b, mk_config(&node_id_hex(2), "b.log", "b.sock")).unwrap(); + + // The generated configs must parse as the lib-level Config type. + let _: tines::config::Config = + serde_json::from_str(&std::fs::read_to_string(&cfg_a).unwrap()).unwrap(); + + let mut a_proc = spawn_tines(&cfg_a, &dir); + let mut b_proc = spawn_tines(&cfg_b, &dir); + + let deadline = Instant::now() + Duration::from_secs(10); + let mut a = connect(&dir.join("a.sock"), deadline); + let mut b = connect(&dir.join("b.sock"), deadline); + + let topic = hex("status"); + let payload = hex("ping"); + + // A observes a broadcast. + rpc( + &mut a, + &Request::Observe { + topic: topic.clone(), + payload: payload.clone(), + to: None, + lifetime_ms: None, + }, + ); + match read_response(&mut a) { + Response::Observe { .. } => {} + other => panic!("expected Observe response, got {other:?}"), + } + + // B's belief converges over the bus. + poll_until( + || { + rpc( + &mut b, + &Request::Belief { + topic: topic.clone(), + }, + ); + read_response(&mut b) + }, + |r| matches!(r, Response::Belief { obs: Some(o) } if o.payload == payload), + deadline, + ); + + // B's store listing contains it too. + rpc(&mut b, &Request::List); + match read_response(&mut b) { + Response::List { observations, .. } => { + assert!( + observations + .iter() + .any(|o| o.payload == payload && o.topic == topic), + "observation missing from B's store list: {observations:?}" + ); + } + other => panic!("expected List response, got {other:?}"), + } + + // Both nodes saw the contact change. + for (stream, expected_peer) in [(&mut a, node_id_hex(2)), (&mut b, node_id_hex(1))] { + poll_until( + || { + rpc(stream, &Request::Events); + read_response(stream) + }, + |r| { + matches!(r, Response::Events { contacts, .. } + if contacts.iter().any(|c| c.joined && c.peer == expected_peer)) + }, + deadline, + ); + } + + drop(a); + drop(b); + let _ = a_proc.kill(); + let _ = b_proc.kill(); + let _ = a_proc.wait(); + let _ = b_proc.wait(); +} + +fn hex(bytes: &str) -> String { + bytes + .as_bytes() + .iter() + .map(|b| format!("{b:02x}")) + .collect() +}