From 0b4b5d1b59cf0d47a0835947230babdc82419d44 Mon Sep 17 00:00:00 2001 From: "K. Isom" Date: Wed, 21 Oct 2026 22:36:05 -0700 Subject: [PATCH] fix(tines): bounded, windowed trusted time for the service - B3: HostClock::wall_ms saturates. New `tines::trust` module: wall times outside WALL_FLOOR_MS (2026-01-01Z) ..= WALL_CEILING_MS (+100 years) are never adopted, from a peer or the system clock. - Decision J: `serve` takes `peer_time_window_ms`; PeerTimeWindow opens on the first plausible estimate, keeps the latest (advanced by ticks), and adopts once when the window closes. The scan stops once trusted. - S5: adopt_wall_clock flushes right after set_wall_clock. - S6: `trusted_clock` accepts only "system", Config denies unknown fields, an implausible system clock is logged and left to the peer path, and the docs say to order the unit after time-sync.target. New config key `peer_time_window_ms` (default 10000). - Tests: the observed_at and time_from_peer tests now run the host's own logic through `serve`; smallest_age_wins is replaced by window and latest-estimate tests (the latter ignored until the store returns the latest estimate); forged u64::MAX and below-floor estimates aren't adopted and don't panic the host. Co-Authored-By: Claude Opus 5.5 --- tines/src/clock.rs | 11 +- tines/src/ipc.rs | 91 ++++++---- tines/src/lib.rs | 1 + tines/src/main.rs | 167 ++++++++++++++---- tines/src/trust.rs | 258 ++++++++++++++++++++++++++++ tines/tests/ipc.rs | 420 ++++++++++++++++++++++++++++----------------- 6 files changed, 725 insertions(+), 223 deletions(-) create mode 100644 tines/src/trust.rs diff --git a/tines/src/clock.rs b/tines/src/clock.rs index 1d7f45fd..a7a379b9 100644 --- a/tines/src/clock.rs +++ b/tines/src/clock.rs @@ -80,7 +80,7 @@ impl HostClock { /// (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)) + .map(|wall| wall.saturating_add(now.saturating_sub(self.anchor_tick))) } /// Whether the host has a trusted wall clock (i.e. trust has been @@ -126,6 +126,15 @@ mod tests { ); } + /// B3: a forged wall near `u64::MAX` saturates instead of overflowing + /// (a panic in debug, a wrap to about 1970 in release). + #[test] + fn wall_ms_saturates_instead_of_overflowing() { + let mut clock = HostClock::start(); + clock.anchor_wall(u64::MAX - 5, 0); + assert_eq!(clock.wall_ms(100), Some(u64::MAX)); + } + #[test] fn now_is_zero_immediately_after_start() { let clock = HostClock::start(); diff --git a/tines/src/ipc.rs b/tines/src/ipc.rs index f2a2fdd0..4dc7e0b4 100644 --- a/tines/src/ipc.rs +++ b/tines/src/ipc.rs @@ -42,6 +42,7 @@ use std::time::Duration; use tines_core::{Node, NodeEvent, Store, Transport, TrimPolicy}; use tines_store::LogStore; +use crate::trust::{adopt_wall_clock, PeerTimeWindow}; use crate::HostClock; const MAX_FRAME: usize = 1 << 20; // 1 MiB cap on a single JSON frame @@ -215,43 +216,63 @@ where } } +/// The serve loop's state between polls: the host clock, the peer-time window, +/// and the poll-round count that schedules trim and flush. +struct HostState { + clock: HostClock, + window: PeerTimeWindow, + rounds: u64, +} + // Drive the exchange until no new observations arrive or the per-iteration cap // (`POLL_CAP`) is reached, then — on schedule — trim and flush. // // 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: &mut HostClock, rounds: &mut u64) +fn poll_exchange(node: &mut Node, host: &mut HostState) where T: Transport + 'static, { - while node.poll(clock.now()) > 0 && *rounds < POLL_CAP { - *rounds += 1; + while node.poll(host.clock.now()) > 0 && host.rounds < POLL_CAP { + host.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.peer_time_estimate(now) { - node.set_wall_clock(est, now); - clock.anchor_wall(est, now); - eprintln!("tines: trusted wall clock from peer (estimate {est} ms)"); + // Peer-time trust (spec §9, decision J). Trusted time comes from a local + // source first — `main` anchors a trusted system clock at startup — and + // otherwise from a peer. While untrusted, re-query the store's estimate + // (drawn only from observations stored since this open; quarantined data + // is invisible) on every poll, and let the window pick the latest + // plausible one; adopt it once, when the window closes. With no peer time + // the host stays quarantined: the robot's own fail-safe is outside tines. + // + // The estimate scans the store, O(n) on every poll. That's acceptable + // only because it stops for good once the clock is trusted. + if !host.clock.is_trusted() { + let now = host.clock.now(); + let was_open = host.window.is_open(); + let adopted = host.window.offer(node.peer_time_estimate(now), now); + if !was_open && host.window.is_open() { + eprintln!("tines: peer wall time seen, collecting estimates before adopting"); + } + if let Some(est) = adopted { + match adopt_wall_clock(node, &mut host.clock, est, now) { + Some(report) => eprintln!( + "tines: trusted wall clock from peer (estimate {est} ms, downtime {:?} ms)", + report.downtime_ms + ), + None => eprintln!("tines: peer wall time {est} ms rejected"), + } } } // 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. - if *rounds >= POLL_CAP { - *rounds = 0; - node.trim(&TrimPolicy::default(), clock.now()); - match node.flush(clock.now()) { + if host.rounds >= POLL_CAP { + host.rounds = 0; + node.trim(&TrimPolicy::default(), host.clock.now()); + match node.flush(host.clock.now()) { Ok(()) => {} Err(e) => eprintln!("tines: store flush failed: {e:?}"), } @@ -273,11 +294,18 @@ where /// /// `poll_interval_ms` is the read-timeout cadence that feeds the poll loop; the /// default (20 ms) keeps the exchange responsive without busy-polling. +/// +/// `clock` is anchored already if the host trusts a local clock. Otherwise +/// `serve` takes its wall time from peers: from the first plausible estimate +/// it collects for `peer_time_window_ms` (decision J; default +/// [`DEFAULT_PEER_TIME_WINDOW_MS`](crate::trust::DEFAULT_PEER_TIME_WINDOW_MS)), +/// then adopts the latest, once. pub fn serve( mut node: Node, socket_path: impl AsRef, - mut clock: HostClock, + clock: HostClock, poll_interval_ms: u64, + peer_time_window_ms: u64, ) -> std::io::Result<()> where T: Transport + 'static, @@ -307,7 +335,11 @@ where } }); - let mut rounds = 0u64; + let mut host = HostState { + clock, + window: PeerTimeWindow::new(peer_time_window_ms), + rounds: 0, + }; loop { match rx.recv_timeout(Duration::from_millis(poll_interval_ms.max(1))) { Ok(mut stream) => { @@ -316,11 +348,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, &mut clock, &mut rounds)?; + serve_stream(&mut node, &mut stream, &mut host)?; } // Timeout: run the exchange poll loop instead of blocking. Err(std::sync::mpsc::RecvTimeoutError::Timeout) => { - poll_exchange(&mut node, &mut clock, &mut rounds); + poll_exchange(&mut node, &mut host); } // The accept thread is gone; stop serving. Err(std::sync::mpsc::RecvTimeoutError::Disconnected) => { @@ -335,8 +367,7 @@ where fn serve_stream( node: &mut Node, stream: &mut UnixStream, - clock: &mut HostClock, - rounds: &mut u64, + host: &mut HostState, ) -> io::Result<()> where T: Transport + 'static, @@ -344,10 +375,10 @@ where loop { match read_request(stream) { Ok(request) => { - let now = clock.now(); + let now = host.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 observed_at = host.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 @@ -357,13 +388,13 @@ where // Drive the exchange after every request so a busy client that // connects back-to-back can't starve it (the WouldBlock path // below only fires once the client goes idle). - poll_exchange(node, clock, rounds); + poll_exchange(node, host); } // Peer closed the connection cleanly. Err(ref e) if e.kind() == std::io::ErrorKind::UnexpectedEof => break, // Idle read: drive the exchange rather than blocking forever. Err(ref e) if e.kind() == std::io::ErrorKind::WouldBlock => { - poll_exchange(node, clock, rounds); + poll_exchange(node, host); } // A malformed frame isn't a disconnect: tell the client and keep // serving this connection. The frame can't be identified, so the diff --git a/tines/src/lib.rs b/tines/src/lib.rs index cda26177..1d209200 100644 --- a/tines/src/lib.rs +++ b/tines/src/lib.rs @@ -12,6 +12,7 @@ pub mod clock; pub mod ipc; +pub mod trust; 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 9e096dfc..766ac98c 100644 --- a/tines/src/main.rs +++ b/tines/src/main.rs @@ -7,17 +7,58 @@ //! The node and peer ids are 32 lowercase hex chars (the 16-byte `NodeId`). //! Sequence numbers are seeded from the LogStore's boot counter, so they don't //! repeat across reboots and need no config field. +//! +//! # Config +//! +//! A JSON object; an unknown key is an error, so a typo can't silently fall +//! back to a default. +//! +//! | key | required | meaning | +//! |---|---|---| +//! | `node_id` | yes | this device's `NodeId`, 32 hex chars | +//! | `peer` | yes | the peer's `NodeId`, 32 hex chars | +//! | `log_path` | yes | the LogStore log file | +//! | `socket_path` | yes | the IPC 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 | +//! | `unknown_downtime_age_ms` | no (LogStore default, 24 h) | broadcast age floor when downtime is unknown | +//! +//! # Trusted time +//! +//! With `trusted_clock = "system"` the service adopts the system clock at +//! startup — if it reads a plausible time (2026 to 2126, see +//! [`tines::trust::WALL_FLOOR_MS`]). A clock outside that range, such as a +//! board that booted before NTP synced and reads 1970, is logged loudly and +//! ignored; the service then takes its time from peers as if the option were +//! unset. Peer time is collected for `peer_time_window_ms` from the first +//! plausible estimate, and the latest is adopted, once per run. +//! +//! The system clock is read once, at startup. Under systemd, order the unit +//! after clock sync so the first read is already disciplined: +//! +//! ```ini +//! [Unit] +//! Wants=time-sync.target +//! After=time-sync.target +//! ``` +//! +//! `time-sync.target` is only reached once the clock is synced when a sync +//! service supports it (`systemd-timesyncd` with +//! `systemd-time-wait-sync.service` enabled, or chrony's equivalent). use std::error::Error; use std::path::Path; +use tines::trust::{adopt_wall_clock, system_wall_ms, DEFAULT_PEER_TIME_WINDOW_MS}; use tines::{serve, HostClock}; use tines_bt::{BluerRadio, BtTransport}; use tines_core::{Node, NodeId}; use tines_store::LogStore; -/// On-disk configuration for a running service. -#[derive(serde::Deserialize)] +/// 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, @@ -35,9 +76,14 @@ struct Config { /// 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. + /// peer observation, and, with none available, stays quarantined. Any + /// other value is a config error. #[serde(default)] - trusted_clock: Option, + 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. @@ -49,26 +95,28 @@ struct Config { 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 } -/// 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 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)?)?; + let config: Config = serde_json::from_str(&std::fs::read_to_string(cfg_path)?) + .map_err(|e| format!("invalid config: {e}"))?; let me = parse_node_id(&config.node_id)?; let peer = parse_node_id(&config.peer)?; @@ -81,34 +129,39 @@ fn main() -> Result<(), Box> { None => LogStore::builder(Path::new(&config.log_path)), }; let mut clock = HostClock::start(); - let mut store = builder + let 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 // 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 // carries a `dyn Radio` with no `Send` bound), so it never crosses a thread. - let node = Node::new(me, store.first_seq(), store, transport); + let mut node = Node::new(me, store.first_seq(), store, transport); + + // Trusted wall clock, in order: (1) a system clock the operator trusts + // (`trusted_clock = "system"`), (2) peers' observations, handled by + // `serve`'s poll loop, or (3) stay quarantined until one is observed. This + // is the system-clock branch: adopt the current system time once, so every + // observation then carries an observed_at. An implausible system clock is + // never adopted (review S6): log it and leave it to the peer path. + if config.trusted_clock == Some(TrustedClock::System) { + let now = clock.now(); + match system_wall_ms().and_then(|wall| adopt_wall_clock(&mut node, &mut clock, wall, now)) { + Some(report) => match report.downtime_ms { + Some(ms) => println!("tines: trusted system clock, folded {ms} ms downtime"), + None => println!("tines: trusted system clock, downtime unknown"), + }, + None => eprintln!( + "tines: WARNING: trusted_clock = \"system\" but the system clock is not \ + plausible (before 2026 or after 2126); NOT trusting it. Falling back to \ + peer time; held data stays quarantined until then. Order the service \ + after time-sync.target." + ), + } + } println!( "tines: listening on {} (node {}, exchange poll {} ms)", @@ -121,6 +174,7 @@ fn main() -> Result<(), Box> { Path::new(&config.socket_path), clock, config.poll_interval_ms, + config.peer_time_window_ms, )?; Ok(()) } @@ -155,3 +209,50 @@ 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/trust.rs b/tines/src/trust.rs new file mode 100644 index 00000000..39486f57 --- /dev/null +++ b/tines/src/trust.rs @@ -0,0 +1,258 @@ +//! Trusted wall time for the hosts (spec §9, decisions G and J). +//! +//! A host opens its store before it knows the wall time, and holds what was +//! already in the log in quarantine until it adopts a trusted wall time. The +//! time comes, in order, from: +//! +//! 1. the robot's own trusted source (the service's `trusted_clock = "system"`, +//! or `TinesNode::trust_system_clock` in tines-ros2), read by +//! [`system_wall_ms`]; +//! 2. otherwise, a peer's observation, through `Node::peer_time_estimate`. +//! [`PeerTimeWindow`] collects estimates for a short window from the first +//! plausible one and adopts the latest (decision J); +//! 3. otherwise the host stays quarantined; the robot's own fail-safe is +//! outside tines. +//! +//! Every candidate passes [`is_plausible_wall_ms`] first (1.0 review, B3): a +//! forged `observed_at = u64::MAX`, or a board that booted before NTP and reads +//! 1970, is never adopted. [`adopt_wall_clock`] is the one place a host commits +//! to a wall time: it folds the store, anchors the [`HostClock`], and flushes so +//! a clock marker carrying the wall time reaches disk (review S5). +//! +//! The time is adopted once per run (decision J): neither path corrects it +//! while the host runs. + +use tines_core::{Node, Tick, Transport}; +use tines_store::log::FoldReport; +use tines_store::LogStore; + +use crate::HostClock; + +/// The earliest wall time a host will adopt: 2026-01-01T00:00:00Z, in ms since +/// the Unix epoch. +/// +/// No tines host predates 2026, so an earlier reading is a clock that was +/// never set (a board without an RTC reads near 1970 until NTP syncs) or a +/// forged peer `observed_at`. Adopting it would stamp observations near 1970, +/// and peers would adopt those stamps in turn. +pub const WALL_FLOOR_MS: u64 = 1_767_225_600_000; + +/// The latest wall time a host will adopt: [`WALL_FLOOR_MS`] plus 100 years +/// (36 525 days), about 2126-01-01. +/// +/// A time ahead of the truth is the safe direction (spec §9: the fold +/// over-counts downtime, so things expire early), so the ceiling is not there +/// to catch a modestly fast clock. It catches values that can only be forged +/// or corrupt, such as `u64::MAX`, which would otherwise overflow every +/// `wall + elapsed` downstream. A century is beyond any deployment's life and +/// still far below where `u64` milliseconds overflow. +pub const WALL_CEILING_MS: u64 = WALL_FLOOR_MS + 36_525 * 24 * 60 * 60 * 1000; + +/// The default peer-time window: how long a host keeps collecting peer +/// estimates after the first plausible one before adopting the latest +/// (decision J, about 10 s). +pub const DEFAULT_PEER_TIME_WINDOW_MS: u64 = 10_000; + +/// Whether `wall_ms` (ms since the Unix epoch) lies within +/// [`WALL_FLOOR_MS`]`..=`[`WALL_CEILING_MS`], so a host may adopt it. +pub fn is_plausible_wall_ms(wall_ms: u64) -> bool { + (WALL_FLOOR_MS..=WALL_CEILING_MS).contains(&wall_ms) +} + +/// The system clock in ms since the Unix epoch, if it is plausible. +/// +/// `None` when the system clock reads before the epoch or outside +/// [`is_plausible_wall_ms`] — typically a board that booted before its clock +/// was synced. The caller should then log it and fall back to the peer path, +/// never anchor 0 (review S6). +pub fn system_wall_ms() -> Option { + use std::time::{SystemTime, UNIX_EPOCH}; + let since = SystemTime::now().duration_since(UNIX_EPOCH).ok()?; + let ms = u64::try_from(since.as_millis()).ok()?; + is_plausible_wall_ms(ms).then_some(ms) +} + +/// Adopt `wall` as the trusted wall time at tick `now`, once. +/// +/// Runs the store's fold (`Node::set_wall_clock`), anchors `clock`, and flushes +/// the store so a marker carrying the wall time reaches disk right away (review +/// S5). Returns the fold report, or `None` — changing nothing — if `clock` is +/// already trusted or `wall` isn't plausible ([`is_plausible_wall_ms`]). +/// +/// A failed flush is logged to stderr, not returned: trust is already +/// established in memory, and the next scheduled flush retries. +pub fn adopt_wall_clock( + node: &mut Node, + clock: &mut HostClock, + wall: u64, + now: Tick, +) -> Option +where + T: Transport, +{ + if clock.is_trusted() || !is_plausible_wall_ms(wall) { + return None; + } + let report = node.set_wall_clock(wall, now); + clock.anchor_wall(wall, now); + if let Err(e) = node.flush(now) { + eprintln!("tines: store flush after adopting wall time failed: {e:?}"); + } + Some(report) +} + +/// Decision J's peer-time window. +/// +/// Feed it the store's current estimate on every poll while the host is +/// untrusted ([`offer`](Self::offer)). The first plausible estimate opens the +/// window; while it is open the window keeps the latest plausible estimate, +/// advanced by local ticks so every candidate is compared at the same instant. +/// When the window has run its length it yields that estimate exactly once, +/// for the host to adopt with [`adopt_wall_clock`]. +/// +/// Implausible estimates ([`is_plausible_wall_ms`]) are ignored: they neither +/// open the window nor compete in it. +#[derive(Debug, Clone)] +pub struct PeerTimeWindow { + /// How long the window stays open after the first plausible estimate. + window_ms: u64, + /// The tick of the first plausible estimate; `None` until one arrives. + opened_at: Option, + /// The latest plausible estimate so far, and the tick it refers to. + best: Option<(u64, Tick)>, + /// Set once the window has yielded its estimate. + done: bool, +} + +impl PeerTimeWindow { + /// A window that stays open `window_ms` after the first plausible + /// estimate. `0` adopts the first plausible estimate immediately. + pub fn new(window_ms: u64) -> Self { + Self { + window_ms, + opened_at: None, + best: None, + done: false, + } + } + + /// Offer the store's current estimate (`None` if it has none) at tick + /// `now`. Returns the estimate to adopt, valued at `now`, once the window + /// has run its length; `None` before then and on every call after. + pub fn offer(&mut self, estimate: Option, now: Tick) -> Option { + if self.done { + return None; + } + // Advance the held best to `now`, so it compares fairly with an + // estimate the store values at `now`. A tick that went backwards + // advances nothing. + let held = self + .best + .map(|(wall, at)| wall.saturating_add(now.saturating_sub(at))); + let fresh = estimate.filter(|&e| is_plausible_wall_ms(e)); + let best = match (held, fresh) { + (Some(h), Some(f)) => h.max(f), + (h, f) => h.or(f)?, + }; + self.best = Some((best, now.max(self.best.map_or(now, |(_, at)| at)))); + let opened_at = *self.opened_at.get_or_insert(now); + if now.saturating_sub(opened_at) < self.window_ms { + return None; + } + self.done = true; + // Advancing by ticks can carry a held estimate past the ceiling only + // after a century of uptime; check anyway rather than adopt it. + is_plausible_wall_ms(best).then_some(best) + } + + /// Whether the window has opened (a plausible estimate has arrived) but + /// not yet yielded. + pub fn is_open(&self) -> bool { + self.opened_at.is_some() && !self.done + } +} + +impl Default for PeerTimeWindow { + fn default() -> Self { + Self::new(DEFAULT_PEER_TIME_WINDOW_MS) + } +} + +#[cfg(test)] +mod tests { + use super::*; + + const W: u64 = 1_800_000_000_000; // 2027-01-15, inside the bounds + + #[test] + fn bounds_reject_epoch_and_overflow_values() { + assert!(!is_plausible_wall_ms(0)); + assert!(!is_plausible_wall_ms(1_700_000_000_000)); // 2023 + assert!(!is_plausible_wall_ms(WALL_FLOOR_MS - 1)); + assert!(is_plausible_wall_ms(WALL_FLOOR_MS)); + assert!(is_plausible_wall_ms(W)); + assert!(is_plausible_wall_ms(WALL_CEILING_MS)); + assert!(!is_plausible_wall_ms(WALL_CEILING_MS + 1)); + assert!(!is_plausible_wall_ms(u64::MAX)); + } + + #[test] + fn floor_is_2026_01_01_utc() { + // 56 years from 1970, 14 of them leap years (1972..=2024). + assert_eq!(WALL_FLOOR_MS, (56 * 365 + 14) * 86_400_000); + } + + #[test] + fn window_waits_then_adopts_latest() { + let mut w = PeerTimeWindow::new(1_000); + assert_eq!(w.offer(None, 0), None); + assert!(!w.is_open()); + assert_eq!( + w.offer(Some(W), 100), + None, + "first estimate opens the window" + ); + assert!(w.is_open()); + assert_eq!(w.offer(Some(W + 5_000), 500), None, "a later, larger one"); + assert_eq!(w.offer(Some(W + 10), 900), None, "a smaller one loses"); + // Window closes at 100 + 1000: the best (W+5000 at tick 500) advanced + // to tick 1100. + assert_eq!(w.offer(Some(W + 600), 1_100), Some(W + 5_600)); + assert!(!w.is_open()); + assert_eq!(w.offer(Some(W + 99_999), 2_000), None, "adopts once"); + } + + #[test] + fn held_estimate_advances_with_ticks() { + let mut w = PeerTimeWindow::new(1_000); + assert_eq!(w.offer(Some(W), 0), None); + // The store no longer reports an estimate (e.g. the observation was + // trimmed); the held one still advances and is adopted. + assert_eq!(w.offer(None, 1_000), Some(W + 1_000)); + } + + #[test] + fn implausible_estimates_never_open_or_win() { + let mut w = PeerTimeWindow::new(0); + assert_eq!(w.offer(Some(u64::MAX), 0), None); + assert_eq!(w.offer(Some(1_000), 1), None); + assert!(!w.is_open()); + assert_eq!(w.offer(Some(W), 2), Some(W), "zero window adopts at once"); + } + + #[test] + fn implausible_estimate_mid_window_is_ignored() { + let mut w = PeerTimeWindow::new(100); + assert_eq!(w.offer(Some(W), 0), None); + assert_eq!(w.offer(Some(u64::MAX), 50), None); + assert_eq!(w.offer(Some(u64::MAX), 100), Some(W + 100)); + } + + #[test] + fn tick_going_backwards_does_not_underflow() { + let mut w = PeerTimeWindow::new(10); + assert_eq!(w.offer(Some(W), 1_000), None); + assert_eq!(w.offer(None, 500), None); + assert_eq!(w.offer(None, 1_010), Some(W + 10)); + } +} diff --git a/tines/tests/ipc.rs b/tines/tests/ipc.rs index 0567ce96..f54dce91 100644 --- a/tines/tests/ipc.rs +++ b/tines/tests/ipc.rs @@ -256,7 +256,8 @@ 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, HostClock::start(), 20).unwrap()); + let _handle = + thread::spawn(move || serve(node, serve_path, HostClock::start(), 20, WINDOW_MS).unwrap()); let mut stream = connect(&socket); @@ -304,7 +305,8 @@ 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, HostClock::start(), 20).unwrap()); + let _handle = + thread::spawn(move || serve(node, serve_path, HostClock::start(), 20, WINDOW_MS).unwrap()); let mut stream = connect(&socket); rpc(&mut stream, &Request::Events); @@ -364,7 +366,8 @@ 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, HostClock::start(), 20).unwrap()); + let _handle = + thread::spawn(move || serve(node, serve_path, HostClock::start(), 20, WINDOW_MS).unwrap()); let mut stream = connect(&socket); // A length-prefixed frame whose body is not valid JSON. @@ -461,14 +464,17 @@ impl Transport for SharedInMemory { } } +/// Nodes A and B over one shared in-memory world, and the world. +type Pair = ( + Node, + Node, + Arc>, +); + /// 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, -) { +/// store, bound to ids `[1;16]` (A) and `[2;16]` (B), and the world itself (so a +/// test can inject frames into either inbox). +fn build_pair(dir: &Path) -> Pair { 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"); @@ -482,12 +488,13 @@ fn build_pair( let tb = SharedInMemory { me: b_id, their: a_id, - world, + world: world.clone(), contacted: false, }; ( Node::new(a_id, a_store.first_seq(), a_store, ta), Node::new(b_id, b_store.first_seq(), b_store, tb), + world, ) } @@ -495,12 +502,7 @@ fn build_pair( /// (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, -) { +fn build_pair_with_held(dir: &Path) -> Pair { 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"); @@ -527,12 +529,13 @@ fn build_pair_with_held( let tb = SharedInMemory { me: b_id, their: a_id, - world, + world: world.clone(), contacted: false, }; ( Node::new(a_id, a_store.first_seq(), a_store, ta), Node::new(b_id, b_store.first_seq(), b_store, tb), + world, ) } @@ -540,7 +543,7 @@ fn build_pair_with_held( #[test] fn f2_direct_propagation() { let dir = unique_dir("probe"); - let (mut node_a, mut node_b) = build_pair(&dir); + 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(); @@ -565,16 +568,16 @@ fn f2_direct_propagation() { #[test] fn f2_propagation_between_instances() { let dir = unique_dir("propagate"); - let (node_a, node_b) = build_pair(&dir); + 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 _a = thread::spawn(move || serve(node_a, socket_a_thread, clock_a, 20, WINDOW_MS).unwrap()); let socket_b_thread = socket_b.clone(); - let _b = thread::spawn(move || serve(node_b, socket_b_thread, clock_b, 20).unwrap()); + let _b = thread::spawn(move || serve(node_b, socket_b_thread, clock_b, 20, WINDOW_MS).unwrap()); // A observes on its socket. let mut stream_a = connect(&socket_a); @@ -612,162 +615,261 @@ fn f2_propagation_between_instances() { panic!("observation through A never became a belief through B"); } -// Trusted wall time (Task 10b): each test pins the brief's requirement verbatim. +// Trusted wall time (spec §9, decisions G and J; 1.0 review B3). +// +// These run the host's own trust logic: B is a served instance (its node moves +// into `serve`), A is a bare node the test pumps. What B decided is read off +// the `observed_at` of an observation B makes over IPC once it reaches A — +// `serve` stamps it from B's `HostClock`, so it is `None` until B adopts a +// wall time and `adopted + elapsed` after. -/// With a trusted clock, an IPC `Observe` produces an observation carrying -/// `observed_at = Some(_)` (R4: once trusted, every observation is stamped). +/// The peer-time window the served instances use: short, to keep tests fast. +const WINDOW_MS: u64 = 300; + +/// A plausible wall time (2027-01-15), inside the hosts' bounds. +const W: u64 = 1_800_000_000_000; + +/// An observation from A with a forged or chosen `observed_at` and `age_ms`. +/// `Observation` is non-exhaustive, so build it and then set the age. +fn crafted(seq: u64, observed_at: u64, age_ms: u64) -> Observation { + let mut obs = + Observation::broadcast([1; 16], seq, &topic(), b"crafted").with_observed_at(observed_at); + obs.age_ms = age_ms; + obs +} + +/// Deliver `obs` to B's inbox as a broadcast from A. +fn inject(world: &Arc>, obs: &Observation) { + let (a_id, b_id) = ([1u8; 16], [2u8; 16]); + let ev = encode_frame(a_id, OBS_TAG, &(a_id, obs.clone())); + world + .lock() + .unwrap() + .inboxes + .entry(b_id) + .or_default() + .push_back(ev); +} + +/// Serve B on its own thread; returns the socket path. +fn serve_b(node_b: Node, clock: HostClock, dir: &Path) -> PathBuf { + let socket = dir.join("b.sock"); + let path = socket.clone(); + let _b = thread::spawn(move || serve(node_b, path, clock, 5, WINDOW_MS).unwrap()); + socket +} + +/// Pump A until `done(A, t)` holds, or panic after ~20 s. +fn pump_a_until( + node_a: &mut Node, + t: &mut u64, + what: &str, + mut done: impl FnMut(&Node, u64) -> bool, +) { + for _ in 0..10_000 { + let _ = node_a.poll(*t); + if done(node_a, *t) { + return; + } + *t += 2; + thread::sleep(Duration::from_millis(2)); + } + panic!("timed out waiting for {what}"); +} + +/// Make an observation on B over IPC, wait for it to reach A, and return the +/// `observed_at` B stamped it with. +fn b_stamp( + socket_b: &Path, + node_a: &mut Node, + t: &mut u64, + n: u32, +) -> Option { + let mut stream = connect(socket_b); + rpc( + &mut stream, + &Request::Observe { + topic: hex_encode(&topic()), + payload: hex_encode(format!("probe-{n}").as_bytes()), + }, + ); + let id = match read_response(&mut stream) { + Response::Observe { id } => id, + other => panic!("expected Observe, got {other:?}"), + }; + let mut found = None; + pump_a_until(node_a, t, "B's observation to reach A", |a, now| { + a.store().for_each_topic(&topic(), now, &mut |obs| { + if hex_encode(&obs.id()) == id { + found = Some(obs.observed_at); + } + }); + found.is_some() + }); + found.expect("found") +} + +/// Probe B until its observations carry a wall time; return the first one. +fn b_adopted_stamp( + socket_b: &Path, + node_a: &mut Node, + t: &mut u64, +) -> u64 { + for n in 0..50 { + if let Some(at) = b_stamp(socket_b, node_a, t, n) { + return at; + } + thread::sleep(Duration::from_millis(WINDOW_MS / 2)); + } + panic!("B never adopted a wall time"); +} + +/// With a trusted clock, an IPC `Observe` through `serve` produces an +/// observation carrying the host's wall time (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); + let (mut node_a, node_b, _) = build_pair(&dir); + let mut clock_b = HostClock::start(); + clock_b.anchor_wall(W, clock_b.now()); + let socket_b = serve_b(node_b, clock_b, &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)); + let mut t = 0; + let at = b_stamp(&socket_b, &mut node_a, &mut t, 0).expect("stamped once trusted"); + assert!( + (W..W + 60_000).contains(&at), + "stamp is the anchored wall plus elapsed ticks: {at}" + ); } -/// Without a trusted clock, an IPC `Observe` carries `observed_at = None` (R4: -/// a observation made while quarantined is un-stamped). +/// Without a trusted clock (and no peer time to adopt), an IPC `Observe` +/// through `serve` carries `observed_at = None` (R4: an 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); + let (mut node_a, node_b, _) = build_pair(&dir); + let socket_b = serve_b(node_b, HostClock::start(), &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); + let mut t = 0; + assert_eq!(b_stamp(&socket_b, &mut node_a, &mut t, 0), 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 - .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. +/// Instance B has no trusted time and a log that already holds one +/// quarantined (held) observation addressed to A. A broadcasts an observation +/// carrying a wall time. B's own host logic — not the test — derives an +/// estimate from it, adopts it once the window closes, and stamps its later +/// observations with it. Adopting runs the unknown-downtime fold, which +/// expires B's held addressed observation, so it never reaches A. #[test] fn time_from_peer() { let dir = unique_dir("peer"); - let (mut node_a, mut node_b) = build_pair_with_held(&dir); + let (mut node_a, node_b, _) = build_pair_with_held(&dir); + let socket_b = serve_b(node_b, HostClock::start(), &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 - .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"); + // A observes at tick 0 with a trusted wall time; the exchange carries it + // to B. + node_a + .observe(&topic(), &payload(), Some(W), 0) + .expect("observe"); + let mut t = 0; + let at = b_adopted_stamp(&socket_b, &mut node_a, &mut t); 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 + (W..W + 60_000).contains(&at), + "B's stamp follows A's wall time: {at}" ); + // The held observation was addressed to A; the fold expired it, so it is + // never delivered. + let mut held_seen = false; + node_a.store().for_each_topic(&topic(), t, &mut |obs| { + held_seen |= obs.payload.as_slice() == b"held"; + }); + assert!(!held_seen, "B's held addressed observation reached A"); +} - // 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.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" +/// Decision J's window: the host doesn't adopt the first plausible estimate. +/// A first observation gives an estimate of about `W`; a second, arriving +/// inside the window, gives about `W + 6000`. The host adopts the later, +/// larger one. +#[test] +fn window_adopts_a_later_larger_estimate() { + let dir = unique_dir("window"); + let (mut node_a, node_b, world) = build_pair(&dir); + let socket_b = serve_b(node_b, HostClock::start(), &dir); + + inject(&world, &crafted(1, W - 1_000, 1_000)); // estimate ~ W + thread::sleep(Duration::from_millis(WINDOW_MS / 3)); + inject(&world, &crafted(2, W + 6_000, 0)); // estimate ~ W + 6000 + + let mut t = 0; + let at = b_adopted_stamp(&socket_b, &mut node_a, &mut t); + assert!( + (W + 6_000..W + 66_000).contains(&at), + "B adopted the larger estimate: stamp {at}, expected >= {}", + W + 6_000 ); } + +/// Decision J's selection: of two observations arriving together, the one +/// with the smaller `age_ms` gives the *smaller* estimate (`W` against +/// `W + 6000`). Smallest-age selection would adopt `W`; the host adopts the +/// latest estimate. +#[test] +#[ignore = "needs decision J store change"] +fn latest_estimate_wins_over_smallest_age() { + let dir = unique_dir("latest"); + let (mut node_a, node_b, world) = build_pair(&dir); + let socket_b = serve_b(node_b, HostClock::start(), &dir); + + inject(&world, &crafted(1, W, 0)); // smaller age, estimate ~ W + inject(&world, &crafted(2, W + 5_000, 1_000)); // estimate ~ W + 6000 + + let mut t = 0; + let at = b_adopted_stamp(&socket_b, &mut node_a, &mut t); + assert!( + (W + 6_000..W + 66_000).contains(&at), + "B adopted the latest estimate: stamp {at}, expected >= {}", + W + 6_000 + ); +} + +/// B stores an observation with the given `observed_at`, keeps serving, and +/// never adopts a wall time from it. +fn assert_not_adopted(tag: &str, observed_at: u64) { + let dir = unique_dir(tag); + let (mut node_a, node_b, world) = build_pair(&dir); + let socket_b = serve_b(node_b, HostClock::start(), &dir); + + let bad = crafted(1, observed_at, 0); + inject(&world, &bad); + // B stored it: it forwards it back to A, which doesn't hold it. + let mut t = 0; + pump_a_until( + &mut node_a, + &mut t, + "B to forward the observation", + |a, _| a.store().contains(&bad.id()), + ); + // Give the window ample time to have closed had B accepted it. + thread::sleep(Duration::from_millis(3 * WINDOW_MS)); + assert_eq!( + b_stamp(&socket_b, &mut node_a, &mut t, 0), + None, + "B adopted an implausible wall time from observed_at = {observed_at}" + ); +} + +/// B3: a forged `observed_at = u64::MAX` neither panics the host nor gets +/// adopted (the estimate saturates to `u64::MAX`, above the ceiling). The +/// host still answers IPC afterwards, which `b_stamp` exercises. +#[test] +fn forged_max_observed_at_is_not_adopted() { + assert_not_adopted("forged", u64::MAX); +} + +/// B3: an estimate below the floor (here 2001) is not adopted. +#[test] +fn below_floor_observed_at_is_not_adopted() { + assert_not_adopted("floor", 1_000_000_000_000); +}