diff --git a/src/features/comments.rs b/src/features/comments.rs index 175f82913..2e6414174 100644 --- a/src/features/comments.rs +++ b/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) diff --git a/src/handlers/message.rs b/src/handlers/message.rs index bfb474bc5..f29125189 100644 --- a/src/handlers/message.rs +++ b/src/handlers/message.rs @@ -98,6 +98,9 @@ fn create_chat_lane(client: &Arc) -> 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 diff --git a/src/message/tests.rs b/src/message/tests.rs index 07499f621..29279cb34 100644 --- a/src/message/tests.rs +++ b/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; diff --git a/src/request.rs b/src/request.rs index 927558928..7538ccdd0 100644 --- a/src/request.rs +++ b/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 { diff --git a/src/send/mod.rs b/src/send/mod.rs index 055c80c17..7593db53d 100644 --- a/src/send/mod.rs +++ b/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, extra_stanza_nodes: Vec, 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, pub(crate) request_id: Option, 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, 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)] diff --git a/src/send/tctoken_lifecycle.rs b/src/send/tctoken_lifecycle.rs index 77e9d85f1..cffc2e107 100644 --- a/src/send/tctoken_lifecycle.rs +++ b/src/send/tctoken_lifecycle.rs @@ -33,12 +33,14 @@ impl Client { &self, to: &Jid, extra_nodes: &mut Vec, + 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`). diff --git a/wacore/src/iq/tctoken.rs b/wacore/src/iq/tctoken.rs index 500d9df6a..edd31fb10 100644 --- a/wacore/src/iq/tctoken.rs +++ b/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, 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, + 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)); diff --git a/wacore/src/request.rs b/wacore/src/request.rs index 9b8e9e027..a791822dd 100644 --- a/wacore/src/request.rs +++ b/wacore/src/request.rs @@ -189,10 +189,16 @@ impl RequestUtils { } pub fn generate_message_id(&self, user_jid: Option<&Jid>) -> String { + self.generate_message_id_at(user_jid, crate::time::now_secs_u64()) + } + + /// Same as [`Self::generate_message_id`], but against a caller-supplied + /// second, so a send that already sampled the clock for its own timestamps + /// derives the id from that same instant instead of reading again. + pub fn generate_message_id_at(&self, user_jid: Option<&Jid>, unix_secs: u64) -> String { let mut data = Vec::with_capacity(8 + 20 + 16); - let timestamp = crate::time::now_secs_u64(); - data.extend_from_slice(×tamp.to_be_bytes()); + data.extend_from_slice(&unix_secs.to_be_bytes()); if let Some(jid) = user_jid { data.extend_from_slice(jid.user.as_bytes()); diff --git a/wacore/src/stats.rs b/wacore/src/stats.rs index 91e398499..6e49769b5 100644 --- a/wacore/src/stats.rs +++ b/wacore/src/stats.rs @@ -11,9 +11,10 @@ //! 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. +//! more when that event carries several frames. A new timestamp field here +//! buys a read on the client's hottest path and needs a reader to justify it: +//! one direct message arrives as roughly four transport events (the message, +//! its ack, the receipt, the receipt's ack), so per-event is per-message x4. //! - [`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