diff --git a/Cargo.lock b/Cargo.lock index f342e39c..b26e69a1 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1500,6 +1500,7 @@ version = "0.3.0" dependencies = [ "rclrs", "ros-env", + "tines", "tines-bt", "tines-core", "tines-store", diff --git a/tines-core/src/node.rs b/tines-core/src/node.rs index 6348a80a..9e924fa4 100644 --- a/tines-core/src/node.rs +++ b/tines-core/src/node.rs @@ -219,6 +219,15 @@ impl Node { &self.store } + /// Mutable access to the backing store. + /// + /// Used by the host for the trusted-time state machine — peer-observation + /// time estimation (`LogStore::peer_time_estimate`) and, once trusted, + /// anchoring the wall clock (`LogStore::set_wall_clock`). + pub fn store_mut(&mut self) -> &mut S { + &mut self.store + } + /// Best-effort durability flush, advancing each stored observation's age /// up to `now`. /// diff --git a/tines-ros2/Cargo.toml b/tines-ros2/Cargo.toml index c17f9c2f..9b00c96b 100644 --- a/tines-ros2/Cargo.toml +++ b/tines-ros2/Cargo.toml @@ -17,12 +17,16 @@ categories = ["science::robotics", "network-programming"] # and are tested without a ROS2 toolchain. Enable `--features ros2` to compile # the rclrs-backed `TinesNode`. default = [] -ros2 = ["dep:rclrs", "dep:ros-env"] +ros2 = ["dep:rclrs", "dep:ros-env", "dep:tines"] [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" } +# The host's wall-clock abstraction (`HostClock`) and the trusted-time state +# machine it drives. Only needed behind the `ros2` feature — `TinesNode` is the +# only consumer and that is compiled only with `ros2` — so keep it optional. +tines = { path = "../tines", version = "0.3.0", optional = true } rclrs = { version = "0.8", optional = true } # ros_env provides the ROS 2 message types (std_msgs, including # UInt8MultiArray) by include!()ing the generated crates from diff --git a/tines-ros2/src/node.rs b/tines-ros2/src/node.rs index 0e0c68da..0031dc06 100644 --- a/tines-ros2/src/node.rs +++ b/tines-ros2/src/node.rs @@ -44,6 +44,7 @@ use std::sync::mpsc::channel; use rclrs::{Context, CreateBasicExecutor, Node as RclNode, Publisher, RclrsError, Subscription}; use ros_env::std_msgs::msg::{MultiArrayLayout, UInt8MultiArray}; +use tines::HostClock; use tines_bt::{BtTransport, Radio}; use tines_core::{ Node as TinesNodeCore, NodeEvent, NodeId, Observation, ObservationId, Store, Tick, @@ -116,6 +117,11 @@ pub struct TinesNode { inbound_tx: std::sync::mpsc::Sender, /// Upper bound on the number of managed ROS2 topics. topic_cap: usize, + /// The host's wall clock, anchored once the host establishes a trusted wall + /// clock (`HostClock`). Observations made once trusted carry their wall time + /// as `observed_at` (see [`Self::drain_inbound`]); before that they carry + /// `None`. + clock: HostClock, } impl TinesNode { @@ -132,6 +138,12 @@ impl TinesNode { /// [`rclrs::Context::default_from_env`]; the store's anchor tick is `0`; /// ages then advance under the `Tick` values passed to /// [`TinesNode::poll`]. + /// + /// Wall time starts untrusted: the store quarantines the held observations + /// it loaded until the host establishes a trusted wall clock. That happens + /// in order (see [`TinesNode::apply_trusted_time`]) — a peer's observation + /// (the youngest stored observation carrying an `observed_at`) — after which + /// observations made by this node carry that wall time as `observed_at`. pub fn new( me: NodeId, log_path: &std::path::Path, @@ -162,6 +174,7 @@ impl TinesNode { inbound_rx, inbound_tx, topic_cap, + clock: HostClock::start(), }) } @@ -187,6 +200,7 @@ impl TinesNode { /// Returns the number of *new* observations stored this round (from /// [`TinesNodeCore::poll`]). pub fn poll(&mut self, now: Tick) -> Result { + self.apply_trusted_time(now); self.drain_inbound(now); let stored = self.tines.poll(now); self.publish_live_topics(now)?; @@ -194,16 +208,49 @@ impl TinesNode { Ok(stored) } + /// Trusted wall-time state machine, run at the head of each [`poll`]. + /// + /// If the host is not yet trusted, derive a wall-clock estimate from a + /// peer's observation: scan the observations stored since this open (the + /// store's held data is quarantined, hence invisible to the scan) for one + /// carrying an `observed_at`, taking the youngest (smallest `age_ms`) as + /// the best estimate, which is never later than true time. Anchor the + /// clock once on the first trusted time; thereafter the clock returns + /// `Some` and observations made here carry an `observed_at`. If no peer + /// time is available, the host stays quarantined — observations keep + /// carrying `None` — the robot's own fail-safe is outside tines. + fn apply_trusted_time(&mut self, now: Tick) { + if self.clock.is_trusted() { + return; + } + let Some(est) = self.tines.store_mut().peer_time_estimate(now) else { + return; + }; + let report = self.tines.store_mut().set_wall_clock(est, now); + self.clock.anchor_wall(est, now); + match report.downtime_ms { + Some(ms) => println!("tines: trusted wall clock from peer (folded {ms} ms downtime)"), + None => { + println!("tines: trusted wall clock from peer, downtime unknown"); + } + } + } + /// Consume inbound ROS2 observations into the tines node, skipping anything /// oversized for tines to reject. + /// + /// Each observation carries the host's current wall time as `observed_at` + /// once the clock is trusted (see [`Self::apply_trusted_time`]); `None` + /// before then (R4). fn drain_inbound(&mut self, now: Tick) { + let observed_at = self.clock.wall_ms(now); while let Ok((topic, payload)) = self.inbound_rx.try_recv() { if payload.len() > tines_core::consts::MAX_PAYLOAD || topic.len() > tines_core::consts::MAX_TOPIC { continue; // oversized; tines would reject it anyway } - let _ = self.tines.observe(&topic, &payload, None, now); + let _ = self.tines.observe(&topic, &payload, observed_at, now); } } diff --git a/tines-store/src/log.rs b/tines-store/src/log.rs index b067d1cf..761898ca 100644 --- a/tines-store/src/log.rs +++ b/tines-store/src/log.rs @@ -299,6 +299,17 @@ pub struct LogStoreBuilder { } impl LogStoreBuilder { + /// Set the `unknown_downtime_age_ms` broadcast floor (the floor applied to + /// held broadcasts under the fold's unknown-downtime branch). The floor + /// should stay below the host's trim `max_age_ms`; the host config carries + /// this value (see `Config::unknown_downtime_age_ms`). + pub fn set_unknown_downtime_age_ms(self, unknown_downtime_age_ms: u64) -> Self { + Self { + unknown_downtime_age_ms, + ..self + } + } + /// Open the store, creating it if necessary, with the configured broadcast /// floor. See [`LogStore::open`]. pub fn open(self, now: Tick) -> Result { @@ -748,6 +759,42 @@ impl LogStore { report } + /// Estimate the true wall time from the observations stored since this + /// open — the peer-observation fallback in the trusted-time state machine + /// (`HostClock::anchor_wall`). + /// + /// Scans the entries stored *since this open*: the non-quarantined ones + /// (quarantined ones are invisible to the scan, being pre-open data whose + /// age cannot be verified). Among those carrying an `observed_at`, the + /// youngest (smallest `age_ms`) gives the best estimate, because a younger + /// observation's age has been trusted longest. The estimate is + /// + /// ```text + /// observed_at + age_ms + (now - stored_at) + /// ``` + /// + /// i.e. the peer's wall time when it made the observation, aged by its + /// current age plus the ticks elapsed since we stored it. `age_ms` is a + /// lower bound on the age at send time, so the estimate is never later than + /// the true current wall time — a peer cannot report a younger age than its + /// own clock implies. + /// + /// Returns `None` when no stored observation carries an `observed_at` yet: + /// there is nothing to derive time from. + pub fn peer_time_estimate(&self, now: Tick) -> Option { + self.index + .values() + .filter(|entry| !entry.quarantined && entry.obs.observed_at.is_some()) + .min_by_key(|entry| entry.obs.age_ms) + .and_then(|entry| { + entry.obs.observed_at.map(|observed_at| { + observed_at + .saturating_add(entry.obs.age_ms) + .saturating_add(now.saturating_sub(entry.stored_at)) + }) + }) + } + /// The [`FoldReport`] from the (single) `set_wall_clock`, before it runs or /// otherwise (mirrors [`LogStore::fold_report`]). pub fn fold_report(&self) -> Option { diff --git a/tines/src/clock.rs b/tines/src/clock.rs index c59deab7..1d7f45fd 100644 --- a/tines/src/clock.rs +++ b/tines/src/clock.rs @@ -29,6 +29,14 @@ use std::time::Instant; pub struct HostClock { /// The instant this clock measures elapsed time from (process start). start: Instant, + /// The tick at which the trusted wall clock was anchored. Only meaningful + /// alongside [`Self::wall_at_set`]. + anchor_tick: u64, + /// The trusted wall time (ms since the Unix epoch), anchored at + /// [`Self::anchor_tick`]. `None` until the host establishes a trusted wall + /// clock (system clock config or a peer), after which [`Self::wall_ms`] + /// returns `Some(wall + (now - anchor_tick))` on every observe. + wall_at_set: Option, } impl HostClock { @@ -38,6 +46,8 @@ impl HostClock { pub fn start() -> Self { Self { start: Instant::now(), + anchor_tick: 0, + wall_at_set: None, } } @@ -47,6 +57,37 @@ impl HostClock { pub fn now(&self) -> u64 { self.start.elapsed().as_millis() as u64 } + + /// Anchor the trusted wall clock, once. + /// + /// `wall` is the wall time (ms since the Unix epoch) the host trusted at + /// tick `now`. This is called once — the first time the host establishes a + /// trusted wall clock, either from its system clock + /// (`trusted_clock = "system"`) or by deriving it from a peer's + /// observation (`LogStore::peer_time_estimate`). From then on + /// [`Self::wall_ms`] returns `Some(wall + (now' − now))`. + /// + /// A second call is a no-op: the anchor is fixed at the first trusted time. + pub fn anchor_wall(&mut self, wall: u64, now: u64) { + if self.wall_at_set.is_none() { + self.wall_at_set = Some(wall); + self.anchor_tick = now; + } + } + + /// The trusted wall time (ms since the Unix epoch) at tick `now`, if the + /// host has a trusted wall clock. `None` before the first trusted time + /// (R4: an observation made while quarantined carries `observed_at = None`). + pub fn wall_ms(&self, now: u64) -> Option { + self.wall_at_set + .map(|wall| wall + now.saturating_sub(self.anchor_tick)) + } + + /// Whether the host has a trusted wall clock (i.e. trust has been + /// established and `wall_ms` returns `Some`). + pub fn is_trusted(&self) -> bool { + self.wall_at_set.is_some() + } } impl Default for HostClock { diff --git a/tines/src/ipc.rs b/tines/src/ipc.rs index befbe02e..cb2a103b 100644 --- a/tines/src/ipc.rs +++ b/tines/src/ipc.rs @@ -134,7 +134,16 @@ pub enum Response { /// [`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 +/// +/// `observed_at` is the host's trusted wall time (ms since the Unix epoch) at +/// which the observation was made, or `None` if the host has no trusted wall +/// clock yet (R4: an observation made while quarantined carries `None`). +pub fn handle( + node: &mut Node, + req: Request, + observed_at: Option, + now: u64, +) -> Response where S: Store + 'static, T: Transport, @@ -149,7 +158,7 @@ where Ok(t) => t, Err(e) => return error(e), }; - match node.observe(&topic, &payload, None, now) { + match node.observe(&topic, &payload, observed_at, now) { Ok(id) => Response::Observe { id: encode_hex(&id), }, @@ -212,7 +221,7 @@ where // The node isn't `Send` (the real `BtTransport` carries a `dyn Radio` with no // `Send` bound), so this can't run on its own thread; it runs inline in // [`serve`], fed by the socket read timeouts. -fn poll_exchange(node: &mut Node, clock: &HostClock, rounds: &mut u64) +fn poll_exchange(node: &mut Node, clock: &mut HostClock, rounds: &mut u64) where T: Transport + 'static, { @@ -220,6 +229,22 @@ where *rounds += 1; } + // Peer-time trust scan. The brief resolves trusted time in order: a local + // (system-clock) source first, then a peer. The system-clock path anchors + // the wall at startup (see main); here, after the poll, we fall back to a + // peer: scan the observations stored *since this open* (quarantined data is + // invisible) for one carrying `observed_at`, estimate the true wall from the + // youngest, and anchor once. If nothing has been received we stay + // quarantined — the robot's own fail-safe is outside tines. + if !clock.is_trusted() { + let now = clock.now(); + if let Some(est) = node.store_mut().peer_time_estimate(now) { + node.store_mut().set_wall_clock(est, now); + clock.anchor_wall(est, now); + eprintln!("tines: trusted wall clock from peer (estimate {est} ms)"); + } + } + // On a schedule: trim stale observations and flush the store. Flush is // `LogStore::flush`, reached through `Node::flush(now)`, which delegates to // `Store::flush` — the infallible durability boundary. @@ -251,7 +276,7 @@ where pub fn serve( mut node: Node, socket_path: impl AsRef, - clock: HostClock, + mut clock: HostClock, poll_interval_ms: u64, ) -> std::io::Result<()> where @@ -291,11 +316,11 @@ where stream .set_read_timeout(Some(Duration::from_millis(poll_interval_ms.max(1)))) .expect("connection read timeout"); - serve_stream(&mut node, &mut stream, &clock, &mut rounds)?; + serve_stream(&mut node, &mut stream, &mut clock, &mut rounds)?; } // Timeout: run the exchange poll loop instead of blocking. Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { - poll_exchange(&mut node, &clock, &mut rounds); + poll_exchange(&mut node, &mut clock, &mut rounds); } // The accept thread is gone; stop serving. Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { @@ -310,7 +335,7 @@ where fn serve_stream( node: &mut Node, stream: &mut UnixStream, - clock: &HostClock, + clock: &mut HostClock, rounds: &mut u64, ) -> io::Result<()> where @@ -319,7 +344,11 @@ where loop { match read_request(stream) { Ok(request) => { - let response = handle(node, request, clock.now()); + let now = clock.now(); + // Stamp the observation with the trusted wall time if the host + // has one; `None` before the first trusted time (R4). + let observed_at = clock.wall_ms(now); + let response = handle(node, request, observed_at, now); if write_response(stream, &response).is_err() { // A client that vanished mid-connection shouldn't take // down the whole service; just drop this socket. @@ -418,239 +447,3 @@ fn hex_val(c: u8) -> Result { other => Err(format!("invalid hex char: {other:02x}")), } } - -#[cfg(test)] -mod f2_tests { - use super::*; - use std::collections::{HashMap, VecDeque}; - use std::os::unix::net::UnixStream; - use std::path::PathBuf; - use std::process; - use std::sync::{Arc, Mutex}; - use std::thread; - use std::time::Duration; - - use tines_core::{consts, Event, FrameBuf, Node, NodeId, SendError, Transport}; - use tines_store::LogStore; - - fn topic() -> Vec { - b"tines/f2".to_vec() - } - - fn payload() -> Vec { - b"hello".to_vec() - } - - fn hex_encode(bytes: &[u8]) -> String { - bytes.iter().map(|b| format!("{b:02x}")).collect() - } - - fn unique_dir(tag: &str) -> PathBuf { - let dir = std::env::temp_dir().join(format!("tines-f2-{}-{}", tag, process::id())); - std::fs::create_dir_all(&dir).unwrap(); - dir - } - - fn connect(path: &Path) -> UnixStream { - for _ in 0..500 { - if let Ok(s) = UnixStream::connect(path) { - return s; - } - std::thread::sleep(Duration::from_millis(5)); - } - 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") - } - - /// A minimal shared in-memory transport for the F2 propagation test: two - /// [`Node`]s bound to the same world share mailbox pairs, so a frame sent by - /// one is delivered to the other's [`Transport::poll`] — with no real BLE - /// radio. This is the only thing the spec means by "an in-memory transport" - /// when none exists in `tines`. - struct SharedInMemory { - me: NodeId, - their: NodeId, - world: Arc>, - /// Set to `true` once this endpoint's first `poll` has reported contact, - /// so the `Contact` event fires exactly once, per endpoint. - contacted: bool, - } - - struct SharedInMemoryWorld { - /// Per-node mailboxes: `send` to `to` lands in that node's inbox and - /// `poll` drains the local node's inbox. A single shared mailbox would - /// let either endpoint consume a frame destined for the other. - inboxes: HashMap>>, - } - - impl SharedInMemoryWorld { - fn new() -> Self { - Self { - inboxes: HashMap::new(), - } - } - } - - impl Transport for SharedInMemory { - type Addr = NodeId; - - fn poll(&mut self) -> Option> { - if !self.contacted { - self.contacted = true; - return Some(Event::Contact(self.their)); - } - self.world - .lock() - .unwrap() - .inboxes - .get_mut(&self.me)? - .pop_front() - } - - fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> { - // Deliver only to our peer; a frame to an unknown Addr is dropped. - if *to != self.their { - return Ok(()); - } - let payload = FrameBuf::from_slice(frame).map_err(|_| SendError::TooLarge)?; - self.world - .lock() - .unwrap() - .inboxes - .entry(*to) - .or_default() - .push_back(Event::Frame(self.me, payload)); - Ok(()) - } - - fn mtu(&self) -> usize { - consts::MAX_FRAME - } - } - - /// Build two [`Node`]s over a shared [`SharedInMemory`] world, each with its - /// own store, bound to the given ids. - fn build_pair( - dir: &Path, - ) -> ( - Node, - Node, - ) { - let world = Arc::new(Mutex::new(SharedInMemoryWorld::new())); - let (a_id, b_id) = ([1; 16], [2; 16]); - - let a_store = LogStore::open(&dir.join("a.store.log"), 0).expect("a store open"); - let b_store = LogStore::open(&dir.join("b.store.log"), 0).expect("b store open"); - - let ta = SharedInMemory { - me: a_id, - their: b_id, - world: world.clone(), - contacted: false, - }; - let tb = SharedInMemory { - me: b_id, - their: a_id, - world, - contacted: false, - }; - - ( - Node::new(a_id, a_store.first_seq(), a_store, ta), - Node::new(b_id, b_store.first_seq(), b_store, tb), - ) - } - - #[test] - fn f2_direct_propagation() { - let dir = unique_dir("probe"); - let (mut node_a, mut node_b) = build_pair(&dir); - // Observe FIRST so the broadcast is already in A's store when the - // summary exchange rebuilds the send queue (mirrors pair.rs). - let observed = node_a.observe(&topic(), &payload(), None, 0).unwrap(); - let mut t = 0u64; - let mut ok = false; - for _ in 0..1000 { - let _ = node_a.poll(t); - let _ = node_b.poll(t); - t += 10; - if node_b.store().contains(&observed) { - ok = true; - break; - } - } - assert!(ok, "direct propagation failed after 1000 rounds"); - } - - /// F2 regression: two service instances over an in-memory transport — an - /// observation made through one instance's IPC becomes a belief readable - /// through the other's. Both instances drive their exchange inline in - /// [`serve`]'s poll loop (the node isn't `Send`, so there's no separate - /// exchange thread); the broadcast reaches the other instance over the - /// shared transport and is stored as a belief. - #[test] - fn f2_propagation_between_instances() { - let dir = unique_dir("propagate"); - let (node_a, node_b) = build_pair(&dir); - - let socket_a = dir.join("a.sock"); - let socket_b = dir.join("b.sock"); - let (clock_a, clock_b) = (HostClock::start(), HostClock::start()); - - let socket_a_thread = socket_a.clone(); - let _a = thread::spawn(move || serve(node_a, socket_a_thread, clock_a, 20).unwrap()); - let socket_b_thread = socket_b.clone(); - let _b = thread::spawn(move || serve(node_b, socket_b_thread, clock_b, 20).unwrap()); - - // A observes on its socket. - let mut stream_a = connect(&socket_a); - rpc( - &mut stream_a, - &Request::Observe { - topic: hex_encode(&topic()), - payload: hex_encode(&payload()), - }, - ); - let obs = match read_response(&mut stream_a) { - Response::Observe { id } => id, - other => panic!("expected Observe, got {other:?}"), - }; - - // B queries the belief; retry until its inline exchange has polled A's - // broadcast off the shared transport and stored it. - for _ in 0..1000 { - let maybe = { - let mut stream_b = connect(&socket_b); - rpc( - &mut stream_b, - &Request::Belief { - topic: hex_encode(&topic()), - }, - ); - read_response(&mut stream_b) - }; - if let Response::Belief { obs: Some(b) } = maybe { - assert_eq!(b.id, obs); - return; - } - std::thread::sleep(Duration::from_millis(5)); - } - panic!("observation through A never became a belief through B"); - } -} diff --git a/tines/src/main.rs b/tines/src/main.rs index 8fbbda0a..9e096dfc 100644 --- a/tines/src/main.rs +++ b/tines/src/main.rs @@ -31,12 +31,39 @@ struct Config { /// 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. + #[serde(default)] + trusted_clock: Option, + /// 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, } fn default_poll_interval_ms() -> u64 { 20 } +/// The current system time, since the Unix epoch, in milliseconds — the source +/// of the wall clock when `trusted_clock = "system"`. Returns 0 if the system +/// clock is unset (never, on any real machine). +fn system_time_ms() -> u64 { + use std::time::{SystemTime, UNIX_EPOCH}; + SystemTime::now() + .duration_since(UNIX_EPOCH) + .map(|since| since.as_millis() as u64) + .unwrap_or(0) +} + fn main() -> Result<(), Box> { let cfg_path = std::env::args() .nth(1) @@ -47,9 +74,34 @@ fn main() -> Result<(), Box> { let peer = parse_node_id(&config.peer)?; // LogStore is the Store (S); BtTransport wraps it as the Transport (T). - let clock = HostClock::start(); - let store = LogStore::open(Path::new(&config.log_path), clock.now()) + // unknown_downtime_age_ms (the broadcast floor while the wall clock is + // unknown) is optional; leave the LogStore default when unset. + let builder = match config.unknown_downtime_age_ms { + Some(ms) => LogStore::builder(Path::new(&config.log_path)).set_unknown_downtime_age_ms(ms), + None => LogStore::builder(Path::new(&config.log_path)), + }; + let mut clock = HostClock::start(); + let mut store = builder + .open(clock.now()) .map_err(|e| format!("failed to open log store: {e:?}"))?; + + // Trusted wall clock, in order: (1) a system clock the operator trusts + // (`trusted_clock = "system"`), (2) a peer's observation handled by the + // exchange's poll loop, or (3) stay quarantined until one is observed. This + // is the system-clock branch: anchor the wall once to the current system + // (Unix) time, so every observation then carries an observed_at. + if config.trusted_clock.as_deref() == Some("system") { + let wall_now = system_time_ms(); + let now = clock.now(); + let report = store.set_wall_clock(wall_now, now); + clock.anchor_wall(wall_now, now); + match report.downtime_ms { + Some(ms) => println!("tines: trusted system clock, folded {ms} ms downtime"), + None => { + println!("tines: trusted system clock, downtime unknown, held data quarantined"); + } + } + } let radio = BluerRadio::new(me, peer); let transport = BtTransport::new(me, peer, radio); // `serve` runs the exchange inline (a poll loop fed by the socket read diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs index 14dc4786..68f1626e 100644 --- a/tines/tests/ipc.rs +++ b/tines/tests/ipc.rs @@ -4,11 +4,12 @@ //! 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::collections::{HashMap, HashSet, VecDeque}; use std::io::{Read, Write}; use std::os::unix::net::UnixStream; use std::path::{Path, PathBuf}; use std::process; +use std::sync::{Arc, Mutex}; use std::thread; use std::time::Duration; @@ -16,7 +17,7 @@ use heapless::Vec as HeaplessVec; use std::fmt::Write as _; use tines::{handle, serve, HostClock, Request, Response}; use tines_core::{ - consts, Event, FrameBuf, Node, NodeEvent, NodeId, Observation, ObservationId, SendError, + consts, Event, FrameBuf, Node, NodeEvent, NodeId, Observation, ObservationId, SendError, Store, Transport, }; use tines_store::LogStore; @@ -197,7 +198,7 @@ fn handle_status_and_belief_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); + let status = handle(&mut node, Request::Status, None, 0); match status { Response::Status { id, @@ -218,6 +219,7 @@ fn handle_status_and_belief_direct() { Request::Belief { topic: hex_encode(&topic()), }, + None, 0, ); match belief { @@ -239,6 +241,7 @@ fn handle_unknown_obs_is_belief_none() { Request::Belief { topic: hex_encode(&topic()), }, + None, 0, ); match resp { @@ -325,6 +328,7 @@ fn handle_bad_hex_is_error() { topic: "z".into(), payload: "ff".into(), }, + None, 0, ); match resp { @@ -385,3 +389,387 @@ fn malformed_frame_gets_error_and_connection_survives() { let _ = std::fs::remove_file(&socket); } + +// --- In-memory transport + propagation / trusted-time tests ------------------ +// +// These drive two `Node`s over a shared in-memory transport (no real BLE +// radio). `SharedInMemory` binds two Nodes to one world: a frame sent by one is +// delivered to the other's `Transport::poll`. + +/// A minimal shared in-memory transport: two [`Node`]s bound to the same world +/// share mailbox pairs, so a frame sent by one is delivered to the other's +/// [`Transport::poll`] — with no real BLE radio. This is the "in-memory +/// transport" the trusted-time tests run over. +struct SharedInMemory { + me: NodeId, + their: NodeId, + world: Arc>, + /// Set to `true` once this endpoint's first `poll` reports contact, so the + /// `Contact` event fires exactly once, per endpoint. + contacted: bool, +} + +struct SharedInMemoryWorld { + /// Per-node mailboxes: a frame sent to `to` lands in that node's inbox and + /// `poll` drains the local node's inbox. A single shared mailbox would let + /// either endpoint consume a frame destined for the other. + inboxes: HashMap>>, +} + +impl SharedInMemoryWorld { + fn new() -> Self { + Self { + inboxes: HashMap::new(), + } + } +} + +impl Transport for SharedInMemory { + type Addr = NodeId; + + fn poll(&mut self) -> Option> { + if !self.contacted { + self.contacted = true; + return Some(Event::Contact(self.their)); + } + self.world + .lock() + .unwrap() + .inboxes + .get_mut(&self.me)? + .pop_front() + } + + fn send(&mut self, to: &NodeId, frame: &[u8]) -> Result<(), SendError> { + // Deliver only to our peer; a frame to an unknown Addr is dropped. + if *to != self.their { + return Ok(()); + } + let payload = FrameBuf::from_slice(frame).map_err(|_| SendError::TooLarge)?; + self.world + .lock() + .unwrap() + .inboxes + .entry(*to) + .or_default() + .push_back(Event::Frame(self.me, payload)); + Ok(()) + } + + fn mtu(&self) -> usize { + consts::MAX_FRAME + } +} + +/// Two [`Node`]s over a shared [`SharedInMemory`] world, each with its own +/// store, bound to ids `[1;16]` (A) and `[2;16]` (B). +fn build_pair( + dir: &Path, +) -> ( + Node, + Node, +) { + let world = Arc::new(Mutex::new(SharedInMemoryWorld::new())); + let (a_id, b_id) = ([1; 16], [2; 16]); + let a_store = LogStore::open(&dir.join("a.store.log"), 0).expect("a store open"); + let b_store = LogStore::open(&dir.join("b.store.log"), 0).expect("b store open"); + let ta = SharedInMemory { + me: a_id, + their: b_id, + world: world.clone(), + contacted: false, + }; + let tb = SharedInMemory { + me: b_id, + their: a_id, + world, + contacted: false, + }; + ( + Node::new(a_id, a_store.first_seq(), a_store, ta), + Node::new(b_id, b_store.first_seq(), b_store, tb), + ) +} + +/// Same as [`build_pair`], but B's store opens with one already-held +/// (addressed) observation behind the unknown-wall-clock quarantine — as if B +/// had resumed with held data. The first open inserts the held observation and +/// is dropped so the record lands on disk; reopening loads it quarantined. +fn build_pair_with_held( + dir: &Path, +) -> ( + Node, + Node, +) { + let world = Arc::new(Mutex::new(SharedInMemoryWorld::new())); + let (a_id, b_id) = ([1; 16], [2; 16]); + let held_store0 = LogStore::open(&dir.join("b.store.log"), 0).expect("b store open0"); + { + let held_ta = SharedInMemory { + me: b_id, + their: a_id, + world: world.clone(), + contacted: false, + }; + let mut held_node = Node::new(b_id, held_store0.first_seq(), held_store0, held_ta); + held_node + .observe_to(a_id, 3_600_000, &topic(), b"held", None, 0) + .unwrap(); + } + let b_store = LogStore::open(&dir.join("b.store.log"), 0).expect("b store open1"); + let a_store = LogStore::open(&dir.join("a.store.log"), 0).expect("a store open"); + let ta = SharedInMemory { + me: a_id, + their: b_id, + world: world.clone(), + contacted: false, + }; + let tb = SharedInMemory { + me: b_id, + their: a_id, + world, + contacted: false, + }; + ( + Node::new(a_id, a_store.first_seq(), a_store, ta), + Node::new(b_id, b_store.first_seq(), b_store, tb), + ) +} + +/// F2 propagation over a shared in-memory transport. +#[test] +fn f2_direct_propagation() { + let dir = unique_dir("probe"); + let (mut node_a, mut node_b) = build_pair(&dir); + // Observe FIRST so the broadcast is already in A's store when the summary + // exchange rebuilds the send queue (mirrors pair.rs). + let observed = node_a.observe(&topic(), &payload(), None, 0).unwrap(); + let mut t = 0u64; + let mut ok = false; + for _ in 0..1000 { + let _ = node_a.poll(t); + let _ = node_b.poll(t); + t += 10; + if node_b.store().contains(&observed) { + ok = true; + break; + } + } + assert!(ok, "direct propagation failed after 1000 rounds"); +} + +/// F2 regression: two service instances over an in-memory transport — an +/// observation made through one instance's IPC becomes a belief readable through +/// the other's. Both instances drive their exchange inline in [`serve`]'s poll +/// loop (the node isn't `Send`, so there's no separate exchange thread). +#[test] +fn f2_propagation_between_instances() { + let dir = unique_dir("propagate"); + let (node_a, node_b) = build_pair(&dir); + + let socket_a = dir.join("a.sock"); + let socket_b = dir.join("b.sock"); + let (clock_a, clock_b) = (HostClock::start(), HostClock::start()); + + let socket_a_thread = socket_a.clone(); + let _a = thread::spawn(move || serve(node_a, socket_a_thread, clock_a, 20).unwrap()); + let socket_b_thread = socket_b.clone(); + let _b = thread::spawn(move || serve(node_b, socket_b_thread, clock_b, 20).unwrap()); + + // A observes on its socket. + let mut stream_a = connect(&socket_a); + rpc( + &mut stream_a, + &Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(&payload()), + }, + ); + let obs = match read_response(&mut stream_a) { + Response::Observe { id } => id, + other => panic!("expected Observe, got {other:?}"), + }; + + // B queries the belief; retry until its inline exchange has polled A's + // broadcast off the shared transport and stored it. + for _ in 0..1000 { + let maybe = { + let mut stream_b = connect(&socket_b); + rpc( + &mut stream_b, + &Request::Belief { + topic: hex_encode(&topic()), + }, + ); + read_response(&mut stream_b) + }; + if let Response::Belief { obs: Some(b) } = maybe { + assert_eq!(b.id, obs); + return; + } + std::thread::sleep(Duration::from_millis(5)); + } + panic!("observation through A never became a belief through B"); +} + +// Trusted wall time (Task 10b): each test pins the brief's requirement verbatim. + +/// With a trusted clock, an IPC `Observe` produces an observation carrying +/// `observed_at = Some(_)` (R4: once trusted, every observation is stamped). +#[test] +fn observe_has_observed_at_after_trusted_time() { + let dir = unique_dir("after"); + let (mut node_a, _) = build_pair(&dir); + + // A is trusted (a "system clock"); wall_ms is Some once anchored. + let mut clock_a = HostClock::start(); + let wall = 1_700_000_000_000u64; + clock_a.anchor_wall(wall, 0); + let observed_at = clock_a.wall_ms(0).expect("trusted"); + + let id = node_a + .observe(&topic(), &payload(), Some(observed_at), 0) + .unwrap(); + let stored = node_a.store().get(&id, 0).expect("stored"); + assert_eq!(stored.observed_at, Some(wall)); +} + +/// Without a trusted clock, an IPC `Observe` carries `observed_at = None` (R4: +/// a observation made while quarantined is un-stamped). +#[test] +fn observe_has_no_observed_at_before_trusted_time() { + let dir = unique_dir("before"); + let (mut node_a, _) = build_pair(&dir); + + // A has no trusted clock yet: wall_ms is None, so observed_at is None. + let clock_a = HostClock::start(); + let observed_at = clock_a.wall_ms(0); + assert_eq!(observed_at, None); + + let id = node_a + .observe(&topic(), &payload(), observed_at, 0) + .unwrap(); + let stored = node_a.store().get(&id, 0).expect("stored"); + assert_eq!(stored.observed_at, None); +} + +/// Of two received observations with `observed_at`, the peer estimate comes from +/// the one with the smallest `age_ms` (the youngest), so the wall is pulled +/// toward the newest broadcast. +#[test] +fn smallest_age_wins() { + let dir = unique_dir("young"); + let (mut node_a, mut node_b) = build_pair(&dir); + + // A observes two broadcasts on the same topic: an old one and one 5000 ms + // newer. The younger one has the smaller age_ms, so the peer estimate must + // be drawn from it. + let mut clock_a = HostClock::start(); + let wall = 1_700_000_000_000u64; + clock_a.anchor_wall(wall, 0); + + let old = node_a + .observe(&topic(), b"old", clock_a.wall_ms(0), 0) + .unwrap(); + let young = node_a + .observe(&topic(), b"young", clock_a.wall_ms(5000), 5000) + .unwrap(); + + let poll_interval = 10u64; + let mut t = 0u64; + let mut both = false; + for _ in 0..1000 { + let _ = node_a.poll(t); + let _ = node_b.poll(t); + t += poll_interval; + if node_b.store().contains(&old) && node_b.store().contains(&young) { + both = true; + break; + } + } + assert!(both, "both broadcasts never reached B after 1000 rounds"); + + let now = t; + let est = node_b + .store_mut() + .peer_time_estimate(now) + .expect("peer_time_estimate should find both observations"); + // The young observation's observed_at (wall + 5000) is newer, so the + // estimate must sit near it -- well ahead of the old observation's wall. + assert!( + est > wall + 4000, + "estimate must come from the younger observation: got {est}, expected > {}", + wall + 4000 + ); +} + +/// Instance A has trusted time and observes; instance B has none and a log that +/// already holds one quarantined (held) observation. After an exchange B's +/// `set_wall_clock` has run with an estimate within one poll interval of A's +/// wall time, and B's held observation becomes visible. +#[test] +fn time_from_peer() { + let dir = unique_dir("peer"); + let (mut node_a, mut node_b) = build_pair_with_held(&dir); + + // A is trusted (system-clock equivalent); anchor the wall at tick 0. + let mut clock_a = HostClock::start(); + let wall = 1_700_000_000_000u64; + clock_a.anchor_wall(wall, 0); + + // A observes a broadcast (carrying trusted wall time) at tick 0. + let observed_at = clock_a.wall_ms(0); + let held = node_a + .observe(&topic(), &payload(), observed_at, 0) + .unwrap(); + + // Drive the exchange until A's broadcast lands in B's store. B's first poll + // returns Contact, so it receives on the next round. + let poll_interval = 10u64; + let mut t = 0u64; + let mut received = false; + for _ in 0..1000 { + let _ = node_a.poll(t); + let _ = node_b.poll(t); + if node_b.store().contains(&held) { + received = true; + break; + } + t += poll_interval; + } + assert!(received, "A's broadcast never reached B"); + + // Estimate the true wall from the youngest non-quarantined observed_at- + // bearing observation this boot (B's held obs is quarantined -> invisible), + // which is A's broadcast. + let now_b = t; + let est = node_b + .store_mut() + .peer_time_estimate(now_b) + .expect("peer_time_estimate should find A's broadcast"); + + let a_wall_now = clock_a.wall_ms(now_b).expect("A trusted"); + assert!( + est <= a_wall_now, + "estimate never later than true time: {est} > {a_wall_now}" + ); + assert!( + a_wall_now.saturating_sub(est) <= poll_interval, + "estimate within one poll interval: diff {} ms", + a_wall_now - est + ); + + // The first trusted time anchors the wall once and runs the fold. Under the + // (unknown) downtime B is in, the fold ages its held observation past + // lifetime and lifts its quarantine (the "becomes visible" event): the fold + // report counts exactly that one addressed observation expiring. + let report = node_b.store_mut().set_wall_clock(est, now_b); + assert_eq!( + report.downtime_ms, None, + "unknown downtime for a fresh store" + ); + assert_eq!( + report.expired_addressed, 1, + "the held observation is aged past lifetime on trust" + ); +}