diff --git a/src/handlers/message.rs b/src/handlers/message.rs index 5d3854559..bfef15fa0 100644 --- a/src/handlers/message.rs +++ b/src/handlers/message.rs @@ -80,15 +80,15 @@ fn create_chat_lane(client: &Arc) -> ChatLane { log::debug!(target: "MessageQueue", "Stale worker exiting; remaining messages will be redelivered by server"); break; } - let start = wacore::time::now_millis() as u64; + let start = wacore::time::Instant::now(); let client = client_for_worker.clone(); Box::pin(client.handle_incoming_message(msg_node)).await; - let elapsed = (wacore::time::now_millis() as u64).saturating_sub(start); - if elapsed > MAX_MESSAGE_DELAY_MS { + let elapsed = start.elapsed(); + if elapsed.as_millis() as u64 > MAX_MESSAGE_DELAY_MS { warn!( target: "MessageQueue", "Message processing took {:.1}s (MAX_MESSAGE_DELAY is {}s)", - elapsed as f64 / 1000.0, + elapsed.as_secs_f64(), MAX_MESSAGE_DELAY_MS / 1000 ); } diff --git a/src/keepalive.rs b/src/keepalive.rs index 4b5ae7740..4d7186398 100644 --- a/src/keepalive.rs +++ b/src/keepalive.rs @@ -65,19 +65,22 @@ impl Client { debug!(target: "Client/Keepalive", "Sending keepalive ping"); + // wall_rtt_ms feeds the WA Web onClockSkewUpdate formula, which + // mixes start_ms with serverTime — both halves must be wall-clock. + // rtt_monotonic is for the log only. let start_ms = wacore::time::now_millis(); + let rtt_start = wacore::time::Instant::now(); let iq = wacore::iq::keepalive::KeepaliveSpec::with_timeout(KEEP_ALIVE_RESPONSE_DEADLINE) .build_iq(); match self.send_iq(iq).await { Ok(response_node) => { - let end_ms = wacore::time::now_millis(); - let rtt_ms = end_ms - start_ms; - debug!(target: "Client/Keepalive", "Received keepalive pong (RTT: {rtt_ms}ms)"); - // WA Web: onClockSkewUpdate — Math.round((startTime + rtt/2) / 1000 - serverTime) + let rtt_monotonic = rtt_start.elapsed(); + let wall_rtt_ms = wacore::time::now_millis().saturating_sub(start_ms).max(0); + debug!(target: "Client/Keepalive", "Received keepalive pong (RTT: {rtt_monotonic:.2?})"); self.unified_session.update_server_time_offset_with_rtt( response_node.get(), start_ms, - rtt_ms, + wall_rtt_ms, ); KeepaliveResult::Ok } diff --git a/wacore/src/time.rs b/wacore/src/time.rs index c9e54d3e0..f7e777e81 100644 --- a/wacore/src/time.rs +++ b/wacore/src/time.rs @@ -1,18 +1,33 @@ -//! Pluggable time provider. +//! Pluggable time providers. //! -//! By default, uses `chrono::Utc::now()`. Can be overridden globally via -//! [`set_time_provider`] for environments where `std::time::SystemTime` is -//! unavailable (e.g. WASM) or for deterministic testing. +//! Two clocks live here, kept distinct on purpose: +//! +//! - **Wall clock** ([`TimeProvider`], [`now_millis`], [`now_utc`]) — answers +//! "what time is it?". Can jump backwards (NTP sync, manual clock changes, +//! leap-second smearing). Resolution: milliseconds is sufficient for +//! timestamps in stanzas, app-state mutations, and log lines. +//! - **Monotonic clock** ([`MonotonicProvider`], [`Instant`]) — answers +//! "how much time passed?". Never moves backwards; immune to NTP +//! adjustments. Resolution: nanoseconds where the platform supports it +//! (`std::time::Instant` on native; sub-millisecond from `performance.now` +//! in browsers, true ns in Node/WASI when those targets supply a custom +//! provider). +//! +//! Conflating the two — using a wall clock to measure elapsed time — silently +//! corrupts timeouts and latency metrics whenever the system clock is adjusted +//! mid-measurement. That is why `std::time` separates `SystemTime` from +//! `Instant`, and we mirror the split here. use std::sync::OnceLock; -/// Trait for providing the current time. +/// Wall-clock provider. Returns the current Unix time. May move backwards +/// across calls when the system clock is adjusted. pub trait TimeProvider: Send + Sync + 'static { /// Current time as milliseconds since Unix epoch. fn now_millis(&self) -> i64; } -/// Default provider using `chrono`. +/// Default wall-clock provider using `chrono`. struct ChronoTimeProvider; impl TimeProvider for ChronoTimeProvider { @@ -27,8 +42,8 @@ impl TimeProvider for ChronoTimeProvider { static TIME_PROVIDER: OnceLock> = OnceLock::new(); -/// Set a custom time provider. Must be called before any time functions are used. -/// Returns `Err` if a provider has already been set. +/// Set a custom wall-clock provider. Must be called before any time function +/// is used. Returns `Err` if a provider has already been set. pub fn set_time_provider(provider: impl TimeProvider) -> Result<(), &'static str> { TIME_PROVIDER .set(Box::new(provider)) @@ -84,40 +99,152 @@ pub fn from_millis_or_now(ts: i64) -> chrono::DateTime { from_millis(ts).unwrap_or_else(now_utc) } -/// Portable monotonic instant, replacing `std::time::Instant` which is -/// unavailable on `wasm32-unknown-unknown`. -/// -/// Uses `now_millis()` internally — not truly monotonic but sufficient -/// for elapsed-time measurement and timeout tracking. +// --------------------------------------------------------------------------- +// Monotonic clock +// --------------------------------------------------------------------------- + +/// Monotonic-clock provider. Returns nanoseconds since an arbitrary fixed +/// reference; the only guarantee is that successive calls never return a +/// smaller value than a previous one. +pub trait MonotonicProvider: Send + Sync + 'static { + /// Nanoseconds since the provider's reference point. The reference is + /// implementation-defined (may be process start, system boot, etc.) — + /// only differences are meaningful. + fn now_nanos(&self) -> u64; +} + +/// Native default: backed by `std::time::Instant`, which is monotonic and +/// has nanosecond resolution on every supported native platform. +#[cfg(not(target_arch = "wasm32"))] +struct StdMonotonicProvider { + epoch: std::time::Instant, +} + +#[cfg(not(target_arch = "wasm32"))] +impl StdMonotonicProvider { + // The single legitimate call to `std::time::Instant::now()`: this IS the + // default provider backing `wacore::time::Instant`. Everywhere else must + // go through this abstraction — see clippy.toml. + #[allow(clippy::disallowed_methods)] + fn new() -> Self { + Self { + epoch: std::time::Instant::now(), + } + } +} + +#[cfg(not(target_arch = "wasm32"))] +impl MonotonicProvider for StdMonotonicProvider { + fn now_nanos(&self) -> u64 { + // Saturate at u64::MAX (584 years) — well beyond any realistic + // process lifetime. + self.epoch.elapsed().as_nanos().min(u64::MAX as u128) as u64 + } +} + +/// WASM fallback when no platform provider is registered. Derives nanos +/// from the wall clock and clamps to non-decreasing so the trait contract +/// holds across NTP backjumps (the value freezes until the wall clock +/// catches up). Resolution is ms × 1_000_000; embedders should register +/// a real provider via [`set_monotonic_provider`] for sub-ms precision. +#[cfg(target_arch = "wasm32")] +struct WallDerivedMonotonicProvider { + last: std::sync::atomic::AtomicU64, +} + +#[cfg(target_arch = "wasm32")] +impl WallDerivedMonotonicProvider { + const fn new() -> Self { + Self { + last: std::sync::atomic::AtomicU64::new(0), + } + } +} + +#[cfg(target_arch = "wasm32")] +impl MonotonicProvider for WallDerivedMonotonicProvider { + fn now_nanos(&self) -> u64 { + use std::sync::atomic::Ordering; + let raw = (now_millis().max(0) as u64).saturating_mul(1_000_000); + let mut last = self.last.load(Ordering::Relaxed); + loop { + let next = raw.max(last); + match self + .last + .compare_exchange_weak(last, next, Ordering::Relaxed, Ordering::Relaxed) + { + Ok(_) => return next, + Err(observed) => last = observed, + } + } + } +} + +static MONOTONIC_PROVIDER: OnceLock> = OnceLock::new(); + +/// Set a custom monotonic-clock provider. Must be called before any +/// [`Instant`] is captured. Returns `Err` if a provider has already been set. +pub fn set_monotonic_provider(provider: impl MonotonicProvider) -> Result<(), &'static str> { + MONOTONIC_PROVIDER + .set(Box::new(provider)) + .map_err(|_| "monotonic provider already set") +} + +#[inline] +fn now_nanos() -> u64 { + MONOTONIC_PROVIDER + .get_or_init(default_monotonic_provider) + .now_nanos() +} + +#[cfg(not(target_arch = "wasm32"))] +fn default_monotonic_provider() -> Box { + Box::new(StdMonotonicProvider::new()) +} + +#[cfg(target_arch = "wasm32")] +fn default_monotonic_provider() -> Box { + Box::new(WallDerivedMonotonicProvider::new()) +} + +/// Portable monotonic instant. On native targets this wraps `std::time::Instant` +/// (via the default [`MonotonicProvider`]) and exposes nanosecond resolution. +/// On `wasm32` targets the embedder should register a sub-millisecond provider +/// via [`set_monotonic_provider`]; otherwise the fallback derives from the +/// wall clock and quantizes to milliseconds. #[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Hash)] -pub struct Instant(i64); +pub struct Instant(u64); impl Instant { - /// Capture the current instant. + /// Capture the current monotonic instant. #[inline] pub fn now() -> Self { - Self(now_millis()) + Self(now_nanos()) } - /// Duration elapsed since this instant was captured. + /// Duration elapsed since this instant was captured. Returns + /// [`Duration::ZERO`] if the clock somehow reported a smaller value + /// than at capture time — which a well-behaved monotonic provider + /// must never do, but we saturate defensively. #[inline] pub fn elapsed(&self) -> std::time::Duration { - let diff = now_millis().saturating_sub(self.0); - std::time::Duration::from_millis(diff.max(0) as u64) + let now = now_nanos(); + std::time::Duration::from_nanos(now.saturating_sub(self.0)) } - /// Duration from this instant until another (saturating). + /// Duration from `earlier` to `self`. Returns [`Duration::ZERO`] if + /// `earlier` is after `self`. #[inline] pub fn saturating_duration_since(&self, earlier: Instant) -> std::time::Duration { - let diff = self.0.saturating_sub(earlier.0); - std::time::Duration::from_millis(diff.max(0) as u64) + std::time::Duration::from_nanos(self.0.saturating_sub(earlier.0)) } } impl std::ops::Add for Instant { type Output = Instant; fn add(self, rhs: std::time::Duration) -> Self { - Self(self.0.saturating_add(rhs.as_millis() as i64)) + let rhs_nanos: u64 = rhs.as_nanos().min(u64::MAX as u128) as u64; + Self(self.0.saturating_add(rhs_nanos)) } }