perf(send): stamp one message from one clock read

src/features/comments.rs

@@ -120,6 +120,7 @@ impl<'a> Comments<'a> {
                 &result.message_id,
                 &comment_secret,
                 wacore::msg_secret::RetentionClass::Text,
+                crate::send::SendInstant::now(),
             )
             .await;
         Ok(result)

src/handlers/message.rs

@@ -98,6 +98,9 @@ fn create_chat_lane(client: &Arc<Client>) -> ChatLane {
                     log::debug!(target: "MessageQueue", "Stale worker exiting; remaining messages will be redelivered by server");
                     break;
                 }
+                // Two clock reads per message, kept: sampling or gating on lane
+                // backlog would stop reporting the single pathological message
+                // this guard exists to catch.
                 let start = wacore::time::Instant::now();
                 let client = client_for_worker.clone();
                 // Awaited inline (not boxed): the future lives in this

src/message/tests.rs

@@ -11354,6 +11354,7 @@ async fn msmsg_outbound_put_and_inbound_get_match_for_lid_bot() {
             outbound_id,
             &secret,
             wacore::msg_secret::RetentionClass::Bot,
+            crate::send::SendInstant::now(),
         )
         .await;
 

src/request.rs

@@ -183,9 +183,15 @@ impl Client {
     ///
     /// A string containing the generated message ID in the format expected by WhatsApp.
     pub fn generate_message_id(&self) -> String {
+        self.generate_message_id_at(wacore::time::now_secs_u64())
+    }
+
+    /// Same as [`Self::generate_message_id`], but against a caller-supplied
+    /// second (see [`RequestUtils::generate_message_id_at`]).
+    pub(crate) fn generate_message_id_at(&self, unix_secs: u64) -> String {
         let device_snapshot = self.persistence_manager.get_device_snapshot();
         self.get_request_utils()
-            .generate_message_id(device_snapshot.pn.as_ref())
+            .generate_message_id_at(device_snapshot.pn.as_ref(), unix_secs)
     }
 
     fn get_request_utils(&self) -> RequestUtils {

src/send/mod.rs

@@ -1,3 +1,10 @@
+//! Outgoing message pipeline.
+//!
+//! Cost model, clock reads: one per send operation, sampled as `SendInstant`
+//! and carried to every stamp. A new timestamp here should take that instant
+//! rather than read again, or the path silently accumulates reads the way it
+//! had accumulated four.
+
 use crate::client::Client;
 use crate::types::message::EditAttribute;
 use anyhow::anyhow;
@@ -175,6 +182,7 @@ struct DmBranchRequest<'a> {
     to: Jid,
     message: &'a wa::Message,
     request_id: String,
+    sent_at: SendInstant,
     edit: Option<EditAttribute>,
     extra_stanza_nodes: Vec<Node>,
     is_status_addon: bool,
@@ -363,8 +371,35 @@ impl EditOptions {
     }
 }
 
+/// The wall-clock second one send operation is stamped with.
+///
+/// Sampled once where the operation starts and carried down, so the message id,
+/// the biz node, the privacy-token decision and the outbound message secret
+/// describe one instant instead of four reads that can straddle a second
+/// boundary, on a path where a clock read is not always cheap.
+#[derive(Debug, Clone, Copy)]
+pub(crate) struct SendInstant(i64);
+
+impl SendInstant {
+    pub(crate) fn now() -> Self {
+        Self(wacore::time::now_secs())
+    }
+
+    pub(crate) fn unix_secs(self) -> i64 {
+        self.0
+    }
+
+    /// Saturated at 0 for the encodings that carry unsigned seconds.
+    pub(crate) fn unix_secs_u64(self) -> u64 {
+        self.0.max(0) as u64
+    }
+}
+
 #[derive(Default)]
 pub(crate) struct SendPipelineOptions {
+    /// Instant this operation is stamped with, when the caller already sampled
+    /// one. `None` makes [`Client::send_message_impl`] sample its own.
+    pub(crate) sent_at: Option<SendInstant>,
     pub(crate) request_id: Option<String>,
     pub(crate) peer: bool,
     pub(crate) force_key_distribution: bool,
@@ -878,9 +913,10 @@ impl Client {
         let stanza_type_override = options.stanza_type_override;
         let group_metadata_freshness = options.group_metadata_freshness;
         let device_freshness = options.device_freshness;
+        let sent_at = SendInstant::now();
         let request_id = match options.message_id {
             Some(id) => id,
-            None => self.generate_message_id(),
+            None => self.generate_message_id_at(sent_at.unix_secs_u64()),
         };
         // Both paths below consume `to` and `request_id`, so save copies for the result.
         let result = SendResult {
@@ -917,8 +953,7 @@ impl Client {
         }
 
         let (edit, inferred_meta) = infer_stanza_metadata(&message);
-        let now_unix_secs = wacore::time::now_secs_u64();
-        let biz = infer_biz_node(&message, now_unix_secs);
+        let biz = infer_biz_node(&message, sent_at.unix_secs_u64());
 
         let extra_nodes =
             build_extra_stanza_nodes(&to, inferred_meta, biz, options.extra_stanza_nodes);
@@ -930,6 +965,7 @@ impl Client {
             to,
             &message,
             SendPipelineOptions {
+                sent_at: Some(sent_at),
                 request_id: Some(request_id),
                 edit,
                 extra_stanza_nodes: extra_nodes,
@@ -1637,6 +1673,7 @@ impl Client {
         options: SendPipelineOptions,
     ) -> Result<(), anyhow::Error> {
         let SendPipelineOptions {
+            sent_at,
             request_id: request_id_override,
             peer,
             force_key_distribution,
@@ -1647,6 +1684,9 @@ impl Client {
             device_freshness,
             borrowed_message_id,
         } = options;
+        // Callers that already stamped their message hand the instant down; the
+        // rest sample here so the pipeline below still has exactly one.
+        let sent_at = sent_at.unwrap_or_else(SendInstant::now);
         validate_extra_stanza_nodes(&extra_stanza_nodes)?;
         if request_id_override.as_ref().is_some_and(String::is_empty) {
             return Err(SendError::InvalidRequest("message ID must not be empty".into()).into());
@@ -1691,7 +1731,7 @@ impl Client {
         // Generate request ID early (doesn't need lock)
         let request_id = match request_id_override {
             Some(id) => id,
-            None => self.generate_message_id(),
+            None => self.generate_message_id_at(sent_at.unix_secs_u64()),
         };
         // `request_id` is moved into the branch-specific stanza builders below;
         // keep a copy for the post-send messageSecret persistence (the secret
@@ -1732,6 +1772,7 @@ impl Client {
                 to,
                 message,
                 request_id,
+                sent_at,
                 edit,
                 extra_stanza_nodes,
                 is_status_addon,
@@ -1800,6 +1841,7 @@ impl Client {
                     &outbound_id_clone,
                     secret,
                     class,
+                    sent_at,
                 )
                 .await;
             }
@@ -2269,6 +2311,7 @@ impl Client {
             to,
             message,
             request_id,
+            sent_at,
             edit,
             extra_stanza_nodes,
             is_status_addon,
@@ -2443,7 +2486,7 @@ impl Client {
             // path but WA Web does not attach tctokens to them.
             if !to.is_group() && !to.is_newsletter() && !is_status_addon {
                 should_issue_tc_token_after_send = self
-                    .maybe_include_tc_token(&to, &mut extra_stanza_nodes)
+                    .maybe_include_tc_token(&to, &mut extra_stanza_nodes, sent_at)
                     .await;
             }
             if should_issue_tc_token_after_send {
@@ -2502,6 +2545,7 @@ impl Client {
         msg_id: &str,
         secret: &[u8; wacore::reporting_token::MESSAGE_SECRET_SIZE],
         class: wacore::msg_secret::RetentionClass,
+        sent_at: SendInstant,
     ) {
         let policy = self.cache_config.msg_secret_policy;
         if !policy.persists() {
@@ -2512,9 +2556,9 @@ impl Client {
         if policy.bot_only() && class != wacore::msg_secret::RetentionClass::Bot {
             return;
         }
-        // Outbound secrets are minted "now", so the parent event time is the
-        // current clock.
-        let now = wacore::time::now_secs();
+        // Outbound secrets are minted with the parent event, so the send's own
+        // instant IS the parent event time.
+        let now = sent_at.unix_secs();
         let expires_at = wacore::msg_secret::expires_at(
             policy,
             &self.cache_config.msg_secret_retention,
@@ -5328,6 +5372,7 @@ mod tests {
                 "MID_1",
                 &secret,
                 wacore::msg_secret::RetentionClass::Text,
+                SendInstant::now(),
             )
             .await;
         client.msg_secret_buffer.wait_flushed().await;
@@ -5356,6 +5401,7 @@ mod tests {
                 "MID_4",
                 &[2u8; 32],
                 wacore::msg_secret::RetentionClass::Text,
+                SendInstant::now(),
             )
             .await;
         client.msg_secret_buffer.wait_flushed().await;
@@ -5438,6 +5484,7 @@ mod tests {
                 "GROUP_MID",
                 &secret,
                 wacore::msg_secret::RetentionClass::Text,
+                SendInstant::now(),
             )
             .await;
         client.msg_secret_buffer.wait_flushed().await;
@@ -5608,7 +5655,7 @@ mod clock_budget_tests {
     /// 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_WALL_READS: u64 = 1;
     const SEND_MONOTONIC_READS: u64 = 2;
 
     async fn seed_devices(client: &Arc<Client>, user: &str) {
@@ -5720,6 +5767,42 @@ mod clock_budget_tests {
         client.disconnect().await;
     }
 
+    /// One send, one instant: the message id, the biz node, the privacy-token
+    /// decision and the outbound secret are stamped from a single read, so they
+    /// cannot end up describing different seconds for the same message.
+    #[tokio::test(start_paused = true)]
+    async fn a_send_reads_the_clock_once() {
+        let (client, _transport, peer) = warm_send_client().await;
+
+        let base = clock_reads::snapshot();
+        let sent = client.send_text(peer.clone(), "hello").await.expect("send");
+        assert_eq!(
+            clock_reads::since(base).wall,
+            1,
+            "every stamp on the send path must come from the same read"
+        );
+
+        // The outbound secret is one of the stamps, so its presence proves the
+        // measured window really covered that write.
+        client.msg_secret_buffer.wait_flushed().await;
+        let stored = client
+            .persistence_manager
+            .backend()
+            .get_msg_secret(
+                &peer.to_non_ad_string(),
+                &format!("{OWN_PN}@s.whatsapp.net"),
+                &sent.message_id,
+            )
+            .await
+            .expect("msg secret lookup");
+        assert!(
+            stored.is_some(),
+            "the send persisted an outbound secret under the measured instant"
+        );
+
+        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)]

src/send/tctoken_lifecycle.rs

@@ -33,12 +33,14 @@ impl Client {
         &self,
         to: &Jid,
         extra_nodes: &mut Vec<Node>,
+        sent_at: SendInstant,
     ) -> bool {
         use wacore::iq::abprops::web;
         use wacore::iq::tctoken::{
             PrivacyTokenChoice, build_cs_token_node, build_tc_token_node, choose_privacy_token,
-            compute_cs_token, is_tc_token_expired_with, should_send_new_tc_token_with,
+            compute_cs_token, is_tc_token_expired_with_at, should_send_new_tc_token_with_at,
         };
+        let now = sent_at.unix_secs();
 
         // Skip for own JID — no need to send a privacy token to ourselves.
         if self.is_own_jid(to) {
@@ -72,16 +74,17 @@ impl Client {
 
         // Issuance scheduling is independent of the AB props — WA Web's sendTcToken
         // in MsgJob.js fires regardless of whether a token was attached to the stanza.
-        let should_issue_after_send = should_send_new_tc_token_with(
+        let should_issue_after_send = should_send_new_tc_token_with_at(
             existing.as_ref().and_then(|entry| entry.sender_timestamp),
             &tc_config,
+            now,
         );
 
         // Bind the token payloads up front so the match arms encode the choice
         // without re-checking invariants that choose_privacy_token already proved.
         let valid_tc_token: Option<&[u8]> = existing.as_ref().and_then(|entry| {
             (!entry.token.is_empty()
-                && !is_tc_token_expired_with(entry.token_timestamp, &tc_config))
+                && !is_tc_token_expired_with_at(entry.token_timestamp, &tc_config, now))
             .then_some(entry.token.as_slice())
         });
         // cstoken needs both the NCT salt and a resolved account LID (WA Web `D`).

wacore/src/iq/tctoken.rs

@@ -95,13 +95,13 @@ fn unix_now() -> i64 {
 
 /// Check if a tcToken has expired using configurable receiver-side timing.
 pub fn is_tc_token_expired_with(token_timestamp: i64, config: &TcTokenConfig) -> bool {
+    is_tc_token_expired_with_at(token_timestamp, config, unix_now())
+}
+
+/// Same as [`is_tc_token_expired_with`], but against a caller-supplied `now`.
+pub fn is_tc_token_expired_with_at(token_timestamp: i64, config: &TcTokenConfig, now: i64) -> bool {
     let cfg = config.clamped();
-    is_tc_token_expired_at(
-        token_timestamp,
-        unix_now(),
-        cfg.bucket_duration,
-        cfg.num_buckets,
-    )
+    is_tc_token_expired_at(token_timestamp, now, cfg.bucket_duration, cfg.num_buckets)
 }
 
 /// Check if a sender-side timestamp has expired using sender-specific timing.
@@ -140,9 +140,18 @@ fn expiration_cutoff_at(now: i64, bucket_duration: i64, num_buckets: i64) -> i64
 pub fn should_send_new_tc_token_with(
     sender_timestamp: Option<i64>,
     config: &TcTokenConfig,
+) -> bool {
+    should_send_new_tc_token_with_at(sender_timestamp, config, unix_now())
+}
+
+/// Same as [`should_send_new_tc_token_with`], but against a caller-supplied `now`.
+pub fn should_send_new_tc_token_with_at(
+    sender_timestamp: Option<i64>,
+    config: &TcTokenConfig,
+    now: i64,
 ) -> bool {
     let cfg = config.clamped();
-    should_send_new_tc_token_at(sender_timestamp, unix_now(), cfg.sender_bucket_duration)
+    should_send_new_tc_token_at(sender_timestamp, now, cfg.sender_bucket_duration)
 }
 
 fn should_send_new_tc_token_at(
@@ -425,6 +434,35 @@ mod tests {
         assert_eq!(bucket_index(1209600, DUR), 2);
     }
 
+    /// The send path hands one instant to both privacy-token decisions instead
+    /// of letting each read the clock. Pin the boundary they land on.
+    #[test]
+    fn supplied_instant_decides_the_same_bucket_boundary() {
+        let config = TcTokenConfig::default().clamped();
+        let issued = 10 * config.sender_bucket_duration;
+
+        let last_second_of_bucket = issued + config.sender_bucket_duration - 1;
+        assert!(!should_send_new_tc_token_with_at(
+            Some(issued),
+            &config,
+            last_second_of_bucket
+        ));
+        assert!(should_send_new_tc_token_with_at(
+            Some(issued),
+            &config,
+            last_second_of_bucket + 1
+        ));
+
+        let stamped = 10 * config.bucket_duration;
+        let still_valid = stamped + (config.num_buckets - 1) * config.bucket_duration;
+        assert!(!is_tc_token_expired_with_at(stamped, &config, still_valid));
+        assert!(is_tc_token_expired_with_at(
+            stamped,
+            &config,
+            still_valid + config.bucket_duration
+        ));
+    }
+
     #[test]
     fn test_should_send_new_tc_token_none() {
         assert!(should_send_new_tc_token_at(None, 1_000_000, DUR));