perf(stats): stop dating every wire frame for a field nothing reads

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.

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]

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.",

src/portable_cache.rs

@@ -365,12 +365,14 @@ where
         K: Borrow<Q>,
         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>,
         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<String, u32> = 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<String, u32> = 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::<String, u32>::builder()
+                .max_capacity(100)
+                .time_to_live(Duration::from_secs(60))
+                .build(),
+            PortableCache::<String, u32>::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<String, String> = PortableCache::builder()