diff --git a/agent_docs/observability.md b/agent_docs/observability.md index f0030c034..f4325b643 100644 --- a/agent_docs/observability.md +++ b/agent_docs/observability.md @@ -32,9 +32,15 @@ chokepoints: transport write — post-noise wire bytes (frame header + AEAD tag included). - **Received**: the read loop (`node_io.rs`) per `DataReceived` batch. -It also owns the `last_data_sent_ms`/`last_data_received_ms` activity -timestamps the keepalive dead-socket watchdog reads (they were loose fields on -`Client` before). Message-level counters piggyback on the existing +It also owns the activity timestamps the keepalive dead-socket watchdog reads: +`last_data_received_ms` (one clock read per received transport event, plus one +more when that event carries several frames, so a slow drain is not read as +silence) and `first_send_since_recv_ms`, which every frame loads but only the +send that arms or re-arms the anchor spends a clock read on. There +is deliberately no "last send" timestamp: nothing in the core reads one, and it +cost a clock read on every frame written, which is the client's hottest path +and a call out of the module on wasm32/embedded. `frames_sent` answers "is it +still sending?" for free. Message-level counters piggyback on the existing `telemetry::send`/`recv` chokepoints; reconnect attempts are counted in the run loop. VoIP relay sockets pass `None` and are not counted — this is the main WA session socket only. diff --git a/src/client/tests.rs b/src/client/tests.rs index fe0729270..eac43f1ea 100644 --- a/src/client/tests.rs +++ b/src/client/tests.rs @@ -3711,6 +3711,91 @@ async fn stats_snapshot_reflects_counters() { assert_eq!(snap.resends_throttled, 0); } +/// A clock read leaves the module on wasm32/embedded, so the wire path owes a +/// budget: only the send that arms the dead-socket anchor may date itself, and +/// only one stamp may be spent per received transport event. +#[test] +fn wire_bookkeeping_reads_the_clock_only_where_a_value_is_used() { + use wacore::time::clock_reads; + + let stats = wacore::stats::SessionStats::new(); + + let arming = clock_reads::snapshot(); + stats.record_frame_sent(10); + assert_eq!( + clock_reads::since(arming).wall, + 1, + "the send that arms the anchor dates it" + ); + + let armed = clock_reads::snapshot(); + for _ in 0..16 { + stats.record_frame_sent(10); + } + assert_eq!( + clock_reads::since(armed).wall, + 0, + "sends under an already-armed anchor have nothing to date" + ); + + let recv = clock_reads::snapshot(); + stats.mark_recv_activity(); + stats.record_recv_batch(100, 1); + assert_eq!( + clock_reads::since(recv).wall, + 1, + "a single-frame batch is stamped once, at arrival" + ); + + let long_batch = clock_reads::snapshot(); + stats.mark_recv_activity(); + stats.record_recv_batch(100, 4); + assert_eq!( + clock_reads::since(long_batch).wall, + 2, + "a long batch re-stamps on completion so a slow drain is not read as silence" + ); + + let rearm = clock_reads::snapshot(); + stats.record_frame_sent(10); + assert_eq!( + clock_reads::since(rearm).wall, + 1, + "the receive cancelled the anchor, so this send arms it again" + ); +} + +/// Handling a received stanza must not ask for the time: the read loop already +/// stamped arrival, and nothing downstream of it dates anything. +#[tokio::test] +async fn received_stanza_handling_reads_no_clock() { + use wacore::time::clock_reads; + + let client = crate::test_utils::create_test_client().await; + let receipt = || { + to_owned_node( + &NodeBuilder::new("receipt") + .attr("id", "3EB0AABBCCDDEEFF001122") + .attr("from", "5511900000001@s.whatsapp.net") + .attr("t", "1780000000") + .build(), + ) + }; + + client.process_decrypted_node(receipt()).await; + crate::test_utils::wait_for_outbound_tasks(&client).await; + + let base = clock_reads::snapshot(); + client.process_decrypted_node(receipt()).await; + let reads = clock_reads::since(base); + + assert_eq!(reads.wall, 0, "receipt handling reads no wall clock"); + assert_eq!( + reads.monotonic, 0, + "receipt handling reads no monotonic clock" + ); +} + /// memory_report must be callable on a fresh client and internally /// consistent: empty collections report zero entries and zero bytes. #[tokio::test] diff --git a/src/keepalive.rs b/src/keepalive.rs index 1ae5f5abd..24c722358 100644 --- a/src/keepalive.rs +++ b/src/keepalive.rs @@ -7,8 +7,8 @@ use std::sync::Arc; use std::time::Duration; use wacore::iq::spec::IqSpec; use wacore::protocol::keepalive::{ - KEEP_ALIVE_INTERVAL_MAX, KEEP_ALIVE_INTERVAL_MIN, KEEP_ALIVE_RESPONSE_DEADLINE, is_dead_socket, - ms_since, + KEEP_ALIVE_INTERVAL_MAX, KEEP_ALIVE_INTERVAL_MIN, KEEP_ALIVE_RESPONSE_DEADLINE, + is_dead_socket_at, ms_since, ms_since_at, }; #[derive(Debug, PartialEq)] @@ -211,8 +211,9 @@ impl Client { // connection died immediately after. let first_send = self.stats.first_send_since_recv_ms(); let last_recv = self.stats.last_data_received_ms(); - if is_dead_socket(first_send, last_recv) { - let elapsed = ms_since(first_send).unwrap_or(0); + let now = wacore::protocol::keepalive::now_ms(); + if is_dead_socket_at(first_send, last_recv, now) { + let elapsed = ms_since_at(first_send, now).unwrap_or(0); warn!( target: "Client/Keepalive", "No data received for {:.1}s after send (dead socket), forcing reconnect.", diff --git a/src/portable_cache.rs b/src/portable_cache.rs index 6367cd93d..d90a5e53c 100644 --- a/src/portable_cache.rs +++ b/src/portable_cache.rs @@ -365,12 +365,14 @@ where K: Borrow, Q: Hash + Eq + ?Sized, { - let now = self.entry_time(); - // Fast path (no TTI): read lock only, no write needed. if self.tti.is_none() { let guard = self.inner.read().await; let entry = guard.map.get(key)?; + // Read the clock after the lookup: a miss has no timestamp to + // compare, and lookups that miss are a large share of the calls + // (every negative registry probe, every warm-up). + let now = self.entry_time(); if self.is_expired(entry, now) { let owned_key = Self::find_key(&guard, key)?; drop(guard); @@ -388,6 +390,7 @@ where // TTI path: write lock to update last_accessed_at. let mut guard = self.inner.write().await; let entry = guard.map.get_mut(key)?; + let now = self.entry_time(); if self.is_expired(entry, now) { let owned_key = Self::find_key(&guard, key)?; guard.remove_key(&owned_key); @@ -491,10 +494,11 @@ where K: Borrow, Q: Hash + Eq + ?Sized, { - let now = self.entry_time(); let mut guard = self.inner.write().await; let owned_key = Self::find_key(&guard, key)?; let entry = guard.remove_key(&owned_key)?; + // Nothing to date until an entry is actually in hand. + let now = self.entry_time(); if self.is_expired(&entry, now) { None } else { @@ -945,6 +949,78 @@ mod tests { assert_eq!(cache.entry_count(), 0); } + /// Expiry decided against a supplied instant, so the boundary (exactly at + /// the deadline, which counts as expired) is pinned without depending on + /// wall-clock timing. + #[test] + fn expiry_boundary_is_exact_under_a_controlled_clock() { + let ttl = Duration::from_secs(60); + let tti = Duration::from_secs(10); + let cache: PortableCache = PortableCache::builder() + .max_capacity(100) + .time_to_live(ttl) + .time_to_idle(tti) + .build(); + + let inserted = Instant::ZERO + Duration::from_secs(1_000); + let entry = CacheEntry { + value: 1, + inserted_at: inserted, + last_accessed_at: inserted, + seq: 0, + }; + + assert!(!cache.is_expired(&entry, inserted + tti - Duration::from_nanos(1))); + assert!(cache.is_expired(&entry, inserted + tti), "TTI is inclusive"); + + let idle_free: PortableCache = PortableCache::builder() + .max_capacity(100) + .time_to_live(ttl) + .build(); + assert!(!idle_free.is_expired(&entry, inserted + ttl - Duration::from_nanos(1))); + assert!( + idle_free.is_expired(&entry, inserted + ttl), + "TTL is inclusive" + ); + } + + /// A lookup that finds nothing has no timestamp to compare, so it must not + /// pay for one. Every negative registry probe goes through here. + #[tokio::test] + async fn a_miss_does_not_read_the_clock() { + use wacore::time::clock_reads; + + for cache in [ + PortableCache::::builder() + .max_capacity(100) + .time_to_live(Duration::from_secs(60)) + .build(), + PortableCache::::builder() + .max_capacity(100) + .time_to_idle(Duration::from_secs(60)) + .build(), + ] { + cache.insert("present".into(), 1).await; + + let base = clock_reads::snapshot(); + assert!(cache.get("absent").await.is_none()); + assert!(cache.remove("absent").await.is_none()); + assert_eq!( + clock_reads::since(base).monotonic, + 0, + "a miss must not read the monotonic clock" + ); + + let hit = clock_reads::snapshot(); + assert_eq!(cache.get("present").await, Some(1)); + assert_eq!( + clock_reads::since(hit).monotonic, + 1, + "a hit reads once, to decide expiry" + ); + } + } + #[tokio::test] async fn test_ttl_expiry() { let cache: PortableCache = PortableCache::builder() diff --git a/src/send/mod.rs b/src/send/mod.rs index 6350dbb53..055c80c17 100644 --- a/src/send/mod.rs +++ b/src/send/mod.rs @@ -5593,3 +5593,171 @@ mod future_size_tests { drop(f); } } + +#[cfg(test)] +mod clock_budget_tests { + use super::*; + use crate::store::commands::DeviceCommand; + use std::sync::Arc; + use wacore::time::clock_reads; + + const OWN_PN: &str = "15551234001"; + const PEER_PN: &str = "5511900000001"; + const PEER_LID: &str = "100000000000079"; + + /// Budget for one steady-state DM send, in clock reads. On wasm32 and + /// embedded targets every read leaves the module, so this is a real cost of + /// the send path and not just an instruction count. + const SEND_WALL_READS: u64 = 4; + const SEND_MONOTONIC_READS: u64 = 2; + + async fn seed_devices(client: &Arc, user: &str) { + client + .update_device_list(wacore::store::traits::DeviceListRecord { + user: user.into(), + devices: vec![wacore::store::traits::DeviceInfo::new(0, None)], + timestamp: wacore::time::now_secs(), + phash: None, + raw_id: None, + }) + .await + .expect("seed device list"); + } + + /// Registry, LID mapping and Signal sessions already seeded, so a send + /// queries nothing over the wire. + async fn cold_send_client() -> ( + Arc, + Arc, + Jid, + ) { + let (client, transport) = crate::test_utils::create_iq_test_client().await; + client + .persistence_manager + .process_command(DeviceCommand::SetId(Some( + format!("{OWN_PN}@s.whatsapp.net").parse().expect("own pn"), + ))) + .await; + client + .persistence_manager + .process_command(DeviceCommand::SetLid(Some( + "100000000000001@lid".parse().expect("own lid"), + ))) + .await; + + let peer = Jid::pn(PEER_PN); + seed_devices(&client, PEER_PN).await; + seed_devices(&client, OWN_PN).await; + seed_devices(&client, PEER_LID).await; + client + .add_lid_pn_mapping( + PEER_LID, + PEER_PN, + crate::lid_pn_cache::LearningSource::Usync, + ) + .await + .expect("lid mapping"); + crate::test_utils::seed_peer_session(&client, &peer).await; + crate::test_utils::seed_peer_session( + &client, + &format!("{PEER_LID}@lid").parse().expect("lid jid"), + ) + .await; + + (client, transport, peer) + } + + /// [`cold_send_client`] plus a first send, which drains the once-per-peer + /// privacy-token issuance so the next send is steady state. + async fn warm_send_client() -> ( + Arc, + Arc, + Jid, + ) { + let (client, transport, peer) = cold_send_client().await; + client.send_text(peer.clone(), "warm").await.expect("warm"); + // The privacy token is issued off the send path, so wait for its frame + // rather than let it land inside the measured window. + crate::test_utils::poll_until("the privacy token to be issued", || { + transport.sent_count() >= 2 + }) + .await; + crate::test_utils::wait_for_outbound_tasks(&client).await; + (client, transport, peer) + } + + /// Paused time so the unanswered IQs this harness leaves behind do not cost + /// their real timeouts. + #[tokio::test(start_paused = true)] + async fn dm_send_stays_within_its_clock_budget() { + let (client, transport, peer) = warm_send_client().await; + let frames_before = transport.sent_count(); + + let base = clock_reads::snapshot(); + client.send_text(peer, "hello").await.expect("send"); + let reads = clock_reads::since(base); + + assert_eq!( + transport.sent_count() - frames_before, + 1, + "the budget only describes a send that writes exactly one frame" + ); + assert!( + reads.total() > 0, + "a zero count means the flow did not run, not that it got free" + ); + assert!( + reads.wall <= SEND_WALL_READS, + "wall-clock reads per DM send rose to {} (budget {SEND_WALL_READS})", + reads.wall + ); + assert!( + reads.monotonic <= SEND_MONOTONIC_READS, + "monotonic reads per DM send rose to {} (budget {SEND_MONOTONIC_READS})", + reads.monotonic + ); + + client.disconnect().await; + } + + /// The wire timestamp is the one thing the budget must never buy: the + /// privacy-token IQ a first send emits still carries the real second. + #[tokio::test(start_paused = true)] + async fn wire_timestamp_keeps_real_time() { + let (client, transport, peer) = cold_send_client().await; + + let before = wacore::time::now_secs(); + client.send_text(peer, "hello").await.expect("send"); + crate::test_utils::poll_until("the privacy token to be issued", || { + transport.sent_count() >= 2 + }) + .await; + let after = wacore::time::now_secs(); + + let mut seen = None; + for index in 0..transport.sent_count() { + let node = crate::test_utils::decode_sent_iq(&transport, index).await; + let node = node.get(); + if node.attrs().optional_string("xmlns").as_deref() != Some("privacy") { + continue; + } + let token = node + .get_optional_child("tokens") + .and_then(|t| t.get_optional_child("token")) + .expect("the privacy IQ carries a "); + seen = token + .attrs() + .optional_string("t") + .and_then(|t| t.parse::().ok()); + break; + } + + let stamped = seen.expect("a first send issues a privacy token"); + assert!( + (before..=after).contains(&stamped), + "wire timestamp {stamped} outside [{before}, {after}]" + ); + + client.disconnect().await; + } +} diff --git a/src/socket/noise_socket.rs b/src/socket/noise_socket.rs index 8fc7e2e0f..d5380b20c 100644 --- a/src/socket/noise_socket.rs +++ b/src/socket/noise_socket.rs @@ -516,7 +516,7 @@ mod tests { let snap = stats.snapshot(); assert_eq!(snap.frames_sent, sent.len() as u64); assert_eq!(snap.bytes_sent, wire_total as u64); - assert!(snap.last_data_sent_ms > 0); + assert!(stats.first_send_since_recv_ms() > 0); } /// Tests edge cases for buffer sizing diff --git a/src/test_utils.rs b/src/test_utils.rs index b2783b7d8..909880019 100644 --- a/src/test_utils.rs +++ b/src/test_utils.rs @@ -320,11 +320,14 @@ pub(crate) async fn create_iq_test_client() -> ( ) .await; - let noise_socket = crate::socket::NoiseSocket::new( + // Wired to the client's stats like the real socket is, so per-frame + // bookkeeping is part of what tests observe. + let noise_socket = crate::socket::NoiseSocket::with_stats( Arc::new(TokioRuntime), transport.clone() as Arc, NoiseCipher::new(&[0u8; 32]).expect("32-byte key"), NoiseCipher::new(&[0u8; 32]).expect("32-byte key"), + Some(client.stats.clone()), ); *client.noise_socket.lock().await = Some(Arc::new(noise_socket)); client.set_connected_for_test(true); diff --git a/wacore/src/protocol/keepalive.rs b/wacore/src/protocol/keepalive.rs index fdc331723..335cd6069 100644 --- a/wacore/src/protocol/keepalive.rs +++ b/wacore/src/protocol/keepalive.rs @@ -20,11 +20,24 @@ pub const DEAD_SOCKET_TIME: Duration = Duration::from_secs(20); /// Returns the number of milliseconds elapsed since a stored timestamp. /// Returns `None` if the timestamp was never set (value 0). pub fn ms_since(timestamp_ms: u64) -> Option { + ms_since_at(timestamp_ms, now_ms()) +} + +/// Same as [`ms_since`], but against a caller-supplied `now`. +/// +/// The dead-socket branch of the keepalive tick evaluates this and +/// [`is_dead_socket_at`] against one instant instead of reading the clock per +/// predicate, and a test can supply the instant outright. +pub fn ms_since_at(timestamp_ms: u64, now_ms: u64) -> Option { if timestamp_ms == 0 { return None; } - let now = crate::time::now_millis().max(0) as u64; - Some(now.saturating_sub(timestamp_ms)) + Some(now_ms.saturating_sub(timestamp_ms)) +} + +/// The wall clock in the units these predicates compare. +pub fn now_ms() -> u64 { + crate::time::now_millis().max(0) as u64 } /// Checks the dead-socket condition: [`DEAD_SOCKET_TIME`] elapsed since the timer @@ -37,6 +50,12 @@ pub fn ms_since(timestamp_ms: u64) -> Option { /// `SessionStats::first_send_since_recv_ms`, reset to 0 on every receive /// (`parseAndHandleStanza` → `cancel()`). pub fn is_dead_socket(armed_ms: u64, last_received_ms: u64) -> bool { + is_dead_socket_at(armed_ms, last_received_ms, now_ms()) +} + +/// Same as [`is_dead_socket`], but against a caller-supplied `now`, so the +/// decision is testable without depending on the platform clock. +pub fn is_dead_socket_at(armed_ms: u64, last_received_ms: u64, now_ms: u64) -> bool { // Timer not armed (never sent since the last receive). if armed_ms == 0 { return false; @@ -45,7 +64,7 @@ pub fn is_dead_socket(armed_ms: u64, last_received_ms: u64) -> bool { if last_received_ms >= armed_ms { return false; } - ms_since(armed_ms) + ms_since_at(armed_ms, now_ms) .map(|elapsed| elapsed > DEAD_SOCKET_TIME.as_millis() as u64) .unwrap_or(false) } @@ -117,6 +136,35 @@ mod tests { assert!(!is_dead_socket(thirty_ago, one_ago)); } + // -- controlled-clock boundary -- + + /// The deadline itself is not yet dead; one millisecond past it is. Pinned + /// against a supplied `now` so the boundary does not depend on how long the + /// test took to run. + #[test] + fn dead_socket_boundary_is_exact() { + let armed = 1_000_000u64; + let dead_at = armed + DEAD_SOCKET_TIME.as_millis() as u64; + assert!(!is_dead_socket_at(armed, 0, dead_at)); + assert!(is_dead_socket_at(armed, 0, dead_at + 1)); + } + + /// A silent socket is still detected while sends keep flowing: the anchor + /// stays put, so elapsed time is measured from the first unanswered send. + #[test] + fn continued_sends_do_not_hide_a_silent_socket() { + let stats = crate::stats::SessionStats::new(); + stats.record_frame_sent(10); + let armed = stats.first_send_since_recv_ms(); + for _ in 0..100 { + stats.record_frame_sent(10); + } + assert_eq!(stats.first_send_since_recv_ms(), armed); + + let now = armed + DEAD_SOCKET_TIME.as_millis() as u64 + 1; + assert!(is_dead_socket_at(armed, 0, now)); + } + // -- constant sanity -- #[test] diff --git a/wacore/src/stats.rs b/wacore/src/stats.rs index a56c6dd06..91e398499 100644 --- a/wacore/src/stats.rs +++ b/wacore/src/stats.rs @@ -9,6 +9,11 @@ //! Cost model: //! - [`SessionStats`] is always on: one relaxed `fetch_add` per wire frame, //! on a path that already does AEAD crypto plus a transport write. +//! - Clock reads: zero per frame sent while the dead-socket anchor is armed, +//! one on the send that arms it, one per received transport event, plus one +//! more when that event carries several frames. On wasm32/embedded every read +//! leaves the module, so a new timestamp field here buys a read on the +//! client's hottest path and needs a reader to justify it. //! - [`HeapSize`] / memory reports only run when called; unused report code //! is dropped by fat LTO. //! - [`TaskInstrument`] is resolved once at client build: unset leaves the @@ -44,16 +49,22 @@ pub struct SessionStats { /// shedding events; the durability hook is the at-least-once escape hatch. events_dropped: AtomicU64, reconnects: AtomicU64, - /// Timestamp (ms since UNIX epoch) of the last sent WebSocket frame. - /// WA Web: `callStanza` → `deadSocketTimer.onOrBefore(deadSocketTime)`. - last_data_sent_ms: AtomicU64, /// Timestamp (ms since UNIX epoch) of the last received WebSocket data. /// WA Web: `parseAndHandleStanza` → `deadSocketTimer.cancel()`. + /// + /// Kept exact: two decisions measure elapsed time from it, the idle-ping + /// gate (15 s) and the dead-socket check (20 s). Sampling it would need a + /// refresh trigger, and the only one the core has is the keepalive tick, + /// which is coarser (15-30 s) than the gate it feeds. last_data_received_ms: AtomicU64, /// Dead-socket watchdog anchor (WA Web `deadSocketTimer.onOrBefore`): the first /// send since the last receive, so continued traffic can't push the deadline out. /// Treated as stale (and re-armed) once `<= last_data_received_ms`, so a send that /// raced past a receive-reset can't leave a pre-receive value stuck here. + /// + /// The only send-side timestamp: a `last_data_sent_ms` companion cost a + /// clock read per frame written and had no reader, since the watchdog + /// anchors on the first unanswered send and never on the most recent one. first_send_since_recv_ms: AtomicU64, } @@ -83,7 +94,6 @@ pub struct StatsSnapshot { /// Outbound resends dropped by the per-chat rate limiter. Surfaces storm /// chats. pub resends_throttled: u64, - pub last_data_sent_ms: u64, pub last_data_received_ms: u64, } @@ -102,19 +112,25 @@ impl SessionStats { self.bytes_sent .fetch_add(wire_bytes as u64, Ordering::Relaxed); self.frames_sent.fetch_add(1, Ordering::Relaxed); - let now = Self::now_ms(); - self.last_data_sent_ms.store(now, Ordering::Relaxed); // Arm the dead-socket deadline on the FIRST send after a receive (WA Web // `onOrBefore` keeps the earliest deadline; later sends must not push it out). - // Re-arm when the anchor is unset OR stale — a receive landed after it was - // armed (`anchor <= last_received`). Guarding only on `== 0` would let a send - // that captured `now` before a concurrent receive-reset write a pre-receive - // timestamp that then sticks forever (its arm raced past the reset), silently - // disabling detection; the stale check re-arms it on the next send instead. + // Re-arm when the anchor is unset OR stale, i.e. a receive landed after it was + // armed: guarding only on `== 0` would let a send whose arm raced past a + // receive-reset leave a pre-receive timestamp stuck there forever, silently + // disabling detection. let last_recv = self.last_data_received_ms.load(Ordering::Relaxed); let anchor = self.first_send_since_recv_ms.load(Ordering::Relaxed); if anchor == 0 || anchor <= last_recv { - self.first_send_since_recv_ms.store(now, Ordering::Relaxed); + // The plain load above gates the clock read; the arm itself re-checks + // under a CAS, so once an anchor is set a later send cannot overwrite + // it. Two senders that both see it unarmed still resolve by whichever + // CAS lands first, which the serial sender task makes unreachable. + let now = Self::now_ms(); + let _ = self.first_send_since_recv_ms.fetch_update( + Ordering::Relaxed, + Ordering::Relaxed, + |current| (current == 0 || current <= last_recv).then_some(now), + ); } } @@ -180,16 +196,10 @@ impl SessionStats { /// watchdog never reads a previous connection's values. Traffic counters /// are cumulative and survive. pub fn reset_connection_activity(&self) { - self.last_data_sent_ms.store(0, Ordering::Relaxed); self.last_data_received_ms.store(0, Ordering::Relaxed); self.first_send_since_recv_ms.store(0, Ordering::Relaxed); } - #[inline] - pub fn last_data_sent_ms(&self) -> u64 { - self.last_data_sent_ms.load(Ordering::Relaxed) - } - /// The dead-socket watchdog anchor: the first send since the last receive /// (0 when unarmed). Evaluate /// [`is_dead_socket`](crate::protocol::keepalive::is_dead_socket) against this, not the last @@ -219,7 +229,6 @@ impl SessionStats { reconnects: self.reconnects.load(Ordering::Relaxed), reconnect_errors: 0, resends_throttled: 0, - last_data_sent_ms: self.last_data_sent_ms.load(Ordering::Relaxed), last_data_received_ms: self.last_data_received_ms.load(Ordering::Relaxed), } } @@ -740,6 +749,7 @@ mod tests { let stats = SessionStats::new(); stats.record_frame_sent(100); stats.record_frame_sent(50); + assert!(stats.first_send_since_recv_ms() > 0); stats.record_recv_batch(300, 2); stats.record_message_sent(); stats.record_message_received(); @@ -756,7 +766,6 @@ mod tests { assert_eq!(snap.messages_received, 1); assert_eq!(snap.reconnects, 1); assert_eq!(snap.events_dropped, 2); - assert!(snap.last_data_sent_ms > 0); assert!(snap.last_data_received_ms > 0); } @@ -845,7 +854,7 @@ mod tests { stats.reset_connection_activity(); let snap = stats.snapshot(); - assert_eq!(snap.last_data_sent_ms, 0); + assert_eq!(stats.first_send_since_recv_ms(), 0); assert_eq!(snap.last_data_received_ms, 0); assert_eq!(snap.bytes_sent, 10); assert_eq!(snap.bytes_received, 20); diff --git a/wacore/src/time.rs b/wacore/src/time.rs index f5dd803f2..b3328451c 100644 --- a/wacore/src/time.rs +++ b/wacore/src/time.rs @@ -20,6 +20,71 @@ use std::sync::OnceLock; +/// Test-only clock-read accounting, so a budget over the hot path can be +/// asserted instead of estimated. +/// +/// Counting sits at the abstraction boundary, not inside a provider: +/// [`set_time_provider`] is a `OnceLock` that the process's first `now_millis()` +/// fills with the default, so a suite sharing one process cannot install an +/// instrumented provider per test. Counting here also covers the defaults. +#[cfg(feature = "test-util")] +pub mod clock_reads { + use core::cell::Cell; + + std::thread_local! { + static WALL: Cell = const { Cell::new(0) }; + static MONOTONIC: Cell = const { Cell::new(0) }; + } + + /// Reads counted on one thread. + #[derive(Debug, Clone, Copy, PartialEq, Eq)] + pub struct Reads { + /// Reads of the wall clock ([`super::now_millis`] and everything built + /// on it). + pub wall: u64, + /// Reads of the monotonic clock ([`super::Instant::now`], + /// [`super::Instant::elapsed`]). + pub monotonic: u64, + } + + impl Reads { + pub fn total(&self) -> u64 { + self.wall + self.monotonic + } + } + + #[inline] + pub(super) fn bump_wall() { + WALL.with(|c| c.set(c.get().saturating_add(1))); + } + + #[inline] + pub(super) fn bump_monotonic() { + MONOTONIC.with(|c| c.set(c.get().saturating_add(1))); + } + + /// Counters for the calling thread. + /// + /// Per thread, not process-wide, so a measurement stays exact while the + /// rest of the suite runs in parallel. Work handed to a blocking pool (the + /// storage backend) is counted on that thread and stays invisible here. + pub fn snapshot() -> Reads { + Reads { + wall: WALL.with(Cell::get), + monotonic: MONOTONIC.with(Cell::get), + } + } + + /// Reads made on this thread since `base`. + pub fn since(base: Reads) -> Reads { + let now = snapshot(); + Reads { + wall: now.wall.saturating_sub(base.wall), + monotonic: now.monotonic.saturating_sub(base.monotonic), + } + } +} + /// 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 { @@ -84,6 +149,8 @@ pub fn set_time_provider(provider: impl TimeProvider) -> Result<(), &'static str #[cfg(not(target_arch = "wasm32"))] #[inline] pub fn now_millis() -> i64 { + #[cfg(feature = "test-util")] + clock_reads::bump_wall(); TIME_PROVIDER .get_or_init(default_time_provider) .now_millis() @@ -95,6 +162,8 @@ pub fn now_millis() -> i64 { #[cfg(target_arch = "wasm32")] #[inline] pub fn now_millis() -> i64 { + #[cfg(feature = "test-util")] + clock_reads::bump_wall(); match TIME_PROVIDER.get() { Some(provider) => provider.now_millis(), None => UnsetWasmTimeProvider.now_millis(), @@ -251,6 +320,8 @@ pub fn set_monotonic_provider(provider: impl MonotonicProvider) -> Result<(), &' #[inline] fn now_nanos() -> u64 { + #[cfg(feature = "test-util")] + clock_reads::bump_monotonic(); MONOTONIC_PROVIDER .get_or_init(default_monotonic_provider) .now_nanos()