diff --git a/tines-ros2/src/node.rs b/tines-ros2/src/node.rs index 38fc030a..ec571a24 100644 --- a/tines-ros2/src/node.rs +++ b/tines-ros2/src/node.rs @@ -175,6 +175,11 @@ impl TinesNode { /// One bridge tick: consume inbound ROS2 observations, drive the tines /// exchange, publish live topics, and publish any confidence events. /// + /// `now` is a monotonic millisecond [`Tick`]: milliseconds since the host + /// process started (a single `Instant`-based clock, `0` at startup), never + /// wall-clock time. Pass the same clock's value every call so ages and + /// addressed lifetimes advance in real milliseconds. + /// /// Returns the number of *new* observations stored this round (from /// [`TinesNodeCore::poll`]). pub fn poll(&mut self, now: Tick) -> Result { diff --git a/tines/src/clock.rs b/tines/src/clock.rs new file mode 100644 index 00000000..c59deab7 --- /dev/null +++ b/tines/src/clock.rs @@ -0,0 +1,94 @@ +//! The host's single monotonic millisecond clock. +//! +//! [`HostClock`] is a single [`std::time::Instant`] captured at process start. +//! Its [`now`](HostClock::now) returns the milliseconds elapsed since that +//! capture as a [`u64`] tines `Tick`. +//! +//! It is monotonic, so an NTP step (forward or backwards) can't rewind it — +//! ages and addressed lifetimes keep advancing instead of saturating or +//! jumping. And it is expressed in milliseconds, the unit the tines `Tick` +//! expects: an observation with `lifetime_ms = 1000` expires after ~1 s of the +//! host's own ticks, not after ~1000 s (which is what a wall-clock *seconds* +//! clock, the old `now_ms`, produced). +//! +//! There is one clock per process. `LogStore` is opened with `clock.now()` +//! (which is `0` at startup, correct by construction), and the service uses +//! the same clock for every `observe`/`belief` tick, so the store and the +//! service never disagree about what "now" is. +//! +//! Wall-clock time (`SystemTime`, milliseconds since the Unix epoch) is used +//! only for `observed_at`, and only when the host is configured to trust its +//! own clock — never for the monotonic tick. + +use std::time::Instant; + +/// One process-wide monotonic millisecond clock. +/// +/// Copy cheaply (a single [`Instant`]); pass it to whoever needs "now". +#[derive(Debug, Clone)] +pub struct HostClock { + /// The instant this clock measures elapsed time from (process start). + start: Instant, +} + +impl HostClock { + /// Capture the clock at process start. + /// + /// `now()` is `0` immediately after this, and grows monotonically. + pub fn start() -> Self { + Self { + start: Instant::now(), + } + } + + /// Milliseconds elapsed since [`HostClock::start`]. + /// + /// `0` right after [`HostClock::start`]; monotonic thereafter. + pub fn now(&self) -> u64 { + self.start.elapsed().as_millis() as u64 + } +} + +impl Default for HostClock { + fn default() -> Self { + Self::start() + } +} + +#[cfg(test)] +mod tests { + use std::time::Duration; + + use super::HostClock; + + /// The tines `Tick` is milliseconds, not seconds. Over a short real sleep + /// the clock must advance by the size of that sleep (hundreds of ms), + /// never by single digits as a wall-clock *seconds* clock would. Fast and + /// non-flaky: a generous lower bound absorbs CI jitter without a long sleep. + #[test] + fn now_advances_in_milliseconds_not_seconds() { + let clock = HostClock::start(); + let first = clock.now(); + std::thread::sleep(Duration::from_millis(150)); + let second = clock.now(); + let delta = second.saturating_sub(first); + assert!( + delta >= 120, + "HostClock::now must advance in milliseconds: expected >=120 ms over a 150 ms sleep, got {delta} ms", + ); + // And it must be ms since process start, not epoch seconds: no process + // has lived a century, so the absolute value stays well under 60 s of + // millis. This would catch a regression back to `as_secs()`. + assert!( + second < 60_000, + "HostClock::now must be ms since process start, was {second}", + ); + } + + #[test] + fn now_is_zero_immediately_after_start() { + let clock = HostClock::start(); + // No sleep: elapsed is a few microseconds, well under a millisecond. + assert!(clock.now() < 1, "clock starts at 0, was {}", clock.now()); + } +} diff --git a/tines/src/ipc.rs b/tines/src/ipc.rs index 940e55a0..f254b360 100644 --- a/tines/src/ipc.rs +++ b/tines/src/ipc.rs @@ -26,9 +26,11 @@ //! //! # 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. +//! `now` is the tines tick passed to `observe`/`belief`. The service reads it +//! from its own monotonic millisecond [`HostClock`] (an `Instant` captured at +//! process start; see [`crate::clock`]) and ignores any caller-supplied value, +//! so the wire request carries no timestamp. The store is opened with the same +//! clock's `now` (0 at startup), so the service and the store agree on "now". use std::fmt::Write as _; use std::io::{Read, Write}; @@ -38,6 +40,8 @@ use std::sync::{Arc, Mutex}; use tines_core::{Node, Store, Transport}; +use crate::HostClock; + const MAX_FRAME: usize = 1 << 20; // 1 MiB cap on a single JSON frame /// A confidence event handed out by the IPC `events` request. @@ -183,9 +187,18 @@ where /// Serve IPC on `socket_path` until the listener dies or a response write /// fails. Blocks; each accepted connection is served request-by-request. /// +/// `now` for each request comes from `clock`, the host's single monotonic +/// millisecond [`HostClock`] (an `Instant` at process start), so the tick the +/// exchange advances on is in real milliseconds. Open the store with the same +/// clock's `now` (0 at startup) so the service and store agree on "now". +/// /// `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<()> +pub fn serve( + node: Node, + socket_path: impl AsRef, + clock: HostClock, +) -> std::io::Result<()> where S: Store + 'static, T: Transport + 'static, @@ -204,7 +217,7 @@ where loop { match read_request(&mut stream) { Ok(req) => { - let now = now_ms(); + let now = clock.now(); let resp = { // Recover from a poisoned lock so one panicking // handler can't take down the daemon. @@ -273,13 +286,6 @@ fn error(msg: impl std::fmt::Display) -> Response { } } -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 { @@ -312,3 +318,25 @@ fn hex_val(c: u8) -> Result { other => Err(format!("invalid hex char: {other:02x}")), } } + +#[cfg(test)] +mod tests { + use std::time::Duration; + + // Regression for F17: the tines `Tick` is milliseconds. A wall-clock + // *seconds* clock (the old `now_ms`) advances by at most ~1 over a + // short interval, so an observation's `lifetime_ms` would expire ~1000× + // too slowly. This asserts the service clock advances in milliseconds. + #[test] + fn tick_advances_in_milliseconds() { + let clock = crate::HostClock::start(); + let a = clock.now(); + std::thread::sleep(Duration::from_millis(150)); + let b = clock.now(); + let delta = b.saturating_sub(a); + assert!( + delta >= 120, + "tines tick advances in milliseconds: expected >=120 ms over a 150 ms sleep, got {delta} ms", + ); + } +} diff --git a/tines/src/lib.rs b/tines/src/lib.rs index 607196ce..cda26177 100644 --- a/tines/src/lib.rs +++ b/tines/src/lib.rs @@ -10,6 +10,8 @@ //! The IPC protocol is documented in [`ipc`]. #![deny(missing_docs)] +pub mod clock; pub mod ipc; +pub use clock::HostClock; pub use ipc::{handle, serve, BeliefOut, EventOut, Request, Response}; diff --git a/tines/src/main.rs b/tines/src/main.rs index 8cb47948..0acb4323 100644 --- a/tines/src/main.rs +++ b/tines/src/main.rs @@ -11,7 +11,7 @@ use std::error::Error; use std::path::Path; -use tines::serve; +use tines::{serve, HostClock}; use tines_bt::{BluerRadio, BtTransport}; use tines_core::{Node, NodeId}; use tines_store::LogStore; @@ -41,7 +41,8 @@ fn main() -> Result<(), Box> { 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) + let clock = HostClock::start(); + let store = LogStore::new(Path::new(&config.log_path), clock.now(), 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); @@ -52,7 +53,7 @@ fn main() -> Result<(), Box> { config.socket_path, node_hex(me) ); - serve(node, Path::new(&config.socket_path))?; + serve(node, Path::new(&config.socket_path), clock)?; Ok(()) } diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs index cfc87917..1cc6f813 100644 --- a/tines/tests/ipc.rs +++ b/tines/tests/ipc.rs @@ -14,7 +14,7 @@ use std::time::Duration; use heapless::Vec as HeaplessVec; use std::fmt::Write as _; -use tines::{handle, serve, Request, Response}; +use tines::{handle, serve, HostClock, Request, Response}; use tines_core::{ consts, Event, Frame, FrameBuf, Node, NodeId, ObservationId, SendError, Transport, }; @@ -220,7 +220,7 @@ fn observe_and_belief_over_socket() { 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 _handle = thread::spawn(move || serve(node, serve_path, HostClock::start()).unwrap()); let mut stream = connect(&socket); @@ -268,7 +268,7 @@ fn events_over_socket() { node.poll(0); // ack processed -> confidence event fires let serve_path = socket.clone(); - let _handle = thread::spawn(move || serve(node, serve_path).unwrap()); + let _handle = thread::spawn(move || serve(node, serve_path, HostClock::start()).unwrap()); let mut stream = connect(&socket); rpc(&mut stream, &Request::Events); @@ -322,7 +322,7 @@ fn malformed_frame_gets_error_and_connection_survives() { let socket = dir.join("ipc.sock"); let node = build_node([3; 16], &dir); let serve_path = socket.clone(); - let _handle = thread::spawn(move || serve(node, serve_path).unwrap()); + let _handle = thread::spawn(move || serve(node, serve_path, HostClock::start()).unwrap()); let mut stream = connect(&socket); // A length-prefixed frame whose body is not valid JSON.