fix(hosts): one monotonic millisecond clock
This commit is contained in:
@@ -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<usize, TinesNodeError> {
|
||||
|
||||
@@ -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());
|
||||
}
|
||||
}
|
||||
+40
-12
@@ -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<S, T>` may be any concrete store/transport, so the real binary passes
|
||||
/// `Node<LogStore, BtTransport>` and tests pass a mock.
|
||||
pub fn serve<S, T>(node: Node<S, T>, socket_path: impl AsRef<Path>) -> std::io::Result<()>
|
||||
pub fn serve<S, T>(
|
||||
node: Node<S, T>,
|
||||
socket_path: impl AsRef<Path>,
|
||||
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<u8, String> {
|
||||
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",
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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};
|
||||
|
||||
+4
-3
@@ -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<dyn Error + Send + Sync>> {
|
||||
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<dyn Error + Send + Sync>> {
|
||||
config.socket_path,
|
||||
node_hex(me)
|
||||
);
|
||||
serve(node, Path::new(&config.socket_path))?;
|
||||
serve(node, Path::new(&config.socket_path), clock)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
|
||||
+4
-4
@@ -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.
|
||||
|
||||
Reference in New Issue
Block a user