From 05c5ff4b3585c3c6c92793394968fabefd2a360f Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 16:33:16 +0000 Subject: [PATCH 1/4] perf(signal): keep skipped message keys out of the protobuf chain A receiver chain's out-of-order keys lived inside the generated `Chain.message_keys`: 136 bytes of `MessageKey` plus a 32-byte `Bytes` per seed-only key, and `decrypt_snapshot` cloned every one of them, promoting each seed to a shared allocation, on every DM decrypt. They now live beside the chain as `Option>>`, a 40-byte inline seed per key, so the snapshot is a refcount bump per chain and a skip-ahead pays one copy-on-write. The protobuf is reassembled only when the state is encoded, and only for a chain that skipped; a record written with the legacy derived triple keeps its protobuf boxed. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_0172fpxasGTrouFyYH5UGmjN --- .../libsignal/src/protocol/state/session.rs | 306 +++++++++++++++--- 1 file changed, 254 insertions(+), 52 deletions(-) diff --git a/wacore/libsignal/src/protocol/state/session.rs b/wacore/libsignal/src/protocol/state/session.rs index 8c200c99f..8a22620f4 100644 --- a/wacore/libsignal/src/protocol/state/session.rs +++ b/wacore/libsignal/src/protocol/state/session.rs @@ -71,9 +71,108 @@ impl UnacknowledgedPreKeyMessageItems { } } +/// One buffered out-of-order message key of a receiver chain. +/// +/// A modern record persists a skipped key as its 32-byte seed alone, and the +/// generated `MessageKey` it decodes into carries that seed as a `Bytes` next +/// to three empty `Option` slots: 136 bytes inline plus a 32-byte +/// allocation for 36 bytes of information. This is the in-memory form, and it +/// is `Copy`-sized: the seed lives inline and there is nothing to allocate or +/// refcount when the chain is cloned. Only a record written before seeds were +/// persisted, which carries the derived cipher/MAC/IV triple instead, keeps +/// its protobuf, boxed so the common arm stays small. +#[derive(Clone)] +pub(crate) enum SkippedKey { + Seed { index: u32, seed: [u8; 32] }, + Legacy(Box), +} + +impl std::fmt::Debug for SkippedKey { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.debug_struct("SkippedKey") + .field("index", &self.index()) + .finish_non_exhaustive() + } +} + +impl SkippedKey { + /// Adopt a decoded key. A seed-only key becomes the inline arm; anything + /// else, the legacy triple included, keeps its protobuf so it round-trips + /// and fails (or succeeds) at lookup exactly as it did before. + fn from_pb(pb: session_structure::chain::MessageKey) -> Self { + if let (Some(index), Some(seed)) = (pb.index, pb.seed.as_deref()) + && let Ok(seed) = <[u8; 32]>::try_from(seed) + && pb.cipher_key.is_none() + && pb.mac_key.is_none() + && pb.iv.is_none() + { + return Self::Seed { index, seed }; + } + Self::Legacy(Box::new(pb)) + } + + fn from_generator(generator: MessageKeyGenerator) -> Self { + match generator { + MessageKeyGenerator::Seed((seed, index)) => Self::Seed { index, seed }, + MessageKeyGenerator::Serialized(pb) => Self::Legacy(Box::new(pb)), + } + } + + fn into_generator(self) -> std::result::Result { + match self { + Self::Seed { index, seed } => Ok(MessageKeyGenerator::new_from_seed(&seed, index)), + Self::Legacy(pb) => MessageKeyGenerator::from_pb(*pb), + } + } + + fn index(&self) -> Option { + match self { + Self::Seed { index, .. } => Some(*index), + Self::Legacy(pb) => pb.index, + } + } + + fn to_pb(&self) -> session_structure::chain::MessageKey { + match self { + Self::Seed { index, seed } => { + MessageKeyGenerator::new_from_seed(seed, *index).into_pb() + } + Self::Legacy(pb) => (**pb).clone(), + } + } + + /// Heap bytes hanging off this key: only the legacy arm owns any. + fn pointed_bytes(&self) -> usize { + match self { + Self::Seed { .. } => 0, + Self::Legacy(pb) => { + size_of::() + + bytes_field_retained(&pb.cipher_key) + + bytes_field_retained(&pb.mac_key) + + bytes_field_retained(&pb.iv) + + bytes_field_retained(&pb.seed) + } + } + } +} + +/// The skipped keys of one receiver chain. `None` until the chain first +/// skips, so the chains of an in-order conversation own nothing; behind an +/// `Arc` so the decrypt snapshot taken before MAC verification is a refcount +/// bump per chain rather than a copy of the whole backlog, which is what it +/// cost while the keys lived inside the cloned protobuf chain. A skip-ahead +/// during a decrypt pays one copy-on-write against that snapshot, for a +/// chain that skipped anyway. +type SkippedKeys = Option>>; + #[derive(Clone, Debug)] pub struct SessionState { session: SessionStructure, + /// Parallel to `session.receiver_chains`, whose own `message_keys` stay + /// empty in memory: this is the source of truth, reassembled into the + /// protobuf only when the state is encoded (`to_protobuf`). Same trick + /// `SenderKeyState` plays with its backlog. + skipped: Vec, } /// Snapshot of the subset of `SessionState` that the decrypt path @@ -85,6 +184,7 @@ pub struct SessionState { /// Held opaque; restore via `SessionState::restore_decrypt_snapshot`. pub struct DecryptSnapshot { receiver_chains: Vec, + skipped: Vec, root_key: Option>, previous_counter: Option, // Stored as `Option` rather than `MessageField` so the snapshot doesn't @@ -121,8 +221,81 @@ fn write_chain_key(field: &mut Option, key: &[u8]) { } impl SessionState { - pub fn from_session_structure(session: SessionStructure) -> Self { - Self { session } + pub fn from_session_structure(mut session: SessionStructure) -> Self { + let skipped = session + .receiver_chains + .iter_mut() + .map(|chain| { + if chain.message_keys.is_empty() { + return None; + } + let keys: Vec = std::mem::take(&mut chain.message_keys) + .into_iter() + .map(SkippedKey::from_pb) + .collect(); + Some(Arc::new(keys)) + }) + .collect(); + Self { session, skipped } + } + + /// Whether any receiver chain holds a skipped key, i.e. whether encoding + /// this state needs the backlog reassembled into the protobuf. + fn has_skipped_keys(&self) -> bool { + self.skipped + .iter() + .any(|keys| keys.as_ref().is_some_and(|keys| !keys.is_empty())) + } + + /// The protobuf with the skipped keys put back, for encoding. Borrowed + /// when there is nothing to put back, which is every chain of an in-order + /// conversation; a clone with the backlog reassembled otherwise. + fn protobuf(&self) -> std::borrow::Cow<'_, SessionStructure> { + if !self.has_skipped_keys() { + return std::borrow::Cow::Borrowed(&self.session); + } + std::borrow::Cow::Owned(self.to_protobuf()) + } + + /// The protobuf with the skipped keys put back, owned. + pub(crate) fn to_protobuf(&self) -> SessionStructure { + let mut session = self.session.clone(); + Self::fill_skipped(&mut session, &self.skipped); + session + } + + /// [`Self::to_protobuf`] without the clone, for a state being consumed. + fn into_protobuf(mut self) -> SessionStructure { + Self::fill_skipped(&mut self.session, &self.skipped); + self.session + } + + fn fill_skipped(session: &mut SessionStructure, skipped: &[SkippedKeys]) { + for (chain, keys) in session.receiver_chains.iter_mut().zip(skipped) { + if let Some(keys) = keys { + chain.message_keys = keys.iter().map(SkippedKey::to_pb).collect(); + } + } + } + + /// Heap bytes the skipped-key backlog points at: the per-chain slots, + /// then for each chain that skipped, the `Arc` and `Vec` headers, the + /// buffer at its capacity, and whatever a legacy key still boxes. The + /// `Vec` header itself is inline in the state, charged by + /// whoever owns the state. + fn skipped_pointed_bytes(&self) -> usize { + self.skipped.capacity() * size_of::() + + self + .skipped + .iter() + .flatten() + .map(|keys| { + 2 * size_of::() + + size_of::>() + + keys.capacity() * size_of::() + + keys.iter().map(SkippedKey::pointed_bytes).sum::() + }) + .sum::() } /// Capture the mutable-during-decrypt fields so MAC failure can @@ -132,6 +305,7 @@ impl SessionState { pub fn decrypt_snapshot(&self) -> DecryptSnapshot { DecryptSnapshot { receiver_chains: self.session.receiver_chains.clone(), + skipped: self.skipped.clone(), root_key: self.session.root_key.clone(), previous_counter: self.session.previous_counter, sender_chain: self.session.sender_chain.as_option().cloned(), @@ -144,6 +318,7 @@ impl SessionState { /// untouched since they were never modified. pub fn restore_decrypt_snapshot(&mut self, snap: DecryptSnapshot) { self.session.receiver_chains = snap.receiver_chains; + self.skipped = snap.skipped; self.session.root_key = snap.root_key; self.session.previous_counter = snap.previous_counter; self.session.sender_chain = snap.sender_chain.into(); @@ -169,6 +344,7 @@ impl SessionState { alice_base_key: Some(alice_base_key.serialize().to_vec()), ..Default::default() }, + skipped: Vec::new(), } } @@ -393,6 +569,7 @@ impl SessionState { }; self.session.receiver_chains.push(chain); + self.skipped.push(None); // Remove oldest chains if we exceed capacity (MAX_RECEIVER_CHAINS = 5). // Using drain() for consistency, though with only 5 elements the difference is negligible. @@ -406,6 +583,7 @@ impl SessionState { ); let excess = len - consts::MAX_RECEIVER_CHAINS; self.session.receiver_chains.drain(..excess); + self.skipped.drain(..excess); } } @@ -535,12 +713,15 @@ impl SessionState { return Ok(None); }; + let Some(keys) = self.skipped.get_mut(chain_idx).and_then(Option::as_mut) else { + return Ok(None); + }; + // Find the message key index without cloning - let chain = &self.session.receiver_chains[chain_idx]; let mut message_key_position = None; - for (i, m) in chain.message_keys.iter().enumerate() { + for (i, m) in keys.iter().enumerate() { let idx = m - .index + .index() .ok_or(InvalidSessionError("missing message key index"))?; if idx == counter { message_key_position = Some(i); @@ -550,11 +731,10 @@ impl SessionState { if let Some(position) = message_key_position { // swap_remove: lookup is by counter, so slot order is free to - // scramble. - let message_key = self.session.receiver_chains[chain_idx] - .message_keys - .swap_remove(position); - let keys = MessageKeyGenerator::from_pb(message_key).map_err(InvalidSessionError)?; + // scramble. The copy-on-write lands only while a decrypt snapshot + // still shares the backlog. + let message_key = Arc::make_mut(keys).swap_remove(position); + let keys = message_key.into_generator().map_err(InvalidSessionError)?; return Ok(Some(keys)); } @@ -570,38 +750,34 @@ impl SessionState { .get_receiver_chain_index(sender)? .expect("called set_message_keys for a non-existent chain"); - let chain = &mut self.session.receiver_chains[chain_idx]; + let keys = Arc::make_mut(self.skipped[chain_idx].get_or_insert_with(Default::default)); // AMORTIZED EVICTION: Only prune when exceeding MAX + threshold. // This reduces O(n) prunes from every insert to once every PRUNE_THRESHOLD inserts. // The lookup in get_message_keys() does a linear search by counter value, so order // doesn't matter for correctness. - let len = chain.message_keys.len(); + let len = keys.len(); if len > consts::MAX_MESSAGE_KEYS + consts::MESSAGE_KEY_PRUNE_THRESHOLD { let excess = len - consts::MAX_MESSAGE_KEYS; // Evict the oldest keys by counter value, not slot position: // swap_remove (here and in get_message_keys) scrambles slot // order, so the front is not the oldest after the first prune. - let mut counters: Vec = chain - .message_keys - .iter() - .map(|m| m.index.unwrap_or(0)) - .collect(); + let mut counters: Vec = keys.iter().map(|m| m.index().unwrap_or(0)).collect(); let (_, &mut threshold, _) = counters.select_nth_unstable(excess - 1); // The removal ceiling keeps duplicate counters at the threshold // (impossible in a valid session) from evicting extra keys. let mut removed = 0; let mut i = 0; - while i < chain.message_keys.len() && removed < excess { - if chain.message_keys[i].index.unwrap_or(0) <= threshold { - chain.message_keys.swap_remove(i); + while i < keys.len() && removed < excess { + if keys[i].index().unwrap_or(0) <= threshold { + keys.swap_remove(i); removed += 1; } else { i += 1; } } } - chain.message_keys.push(message_keys.into_pb()); + keys.push(SkippedKey::from_generator(message_keys)); Ok(()) } @@ -721,13 +897,13 @@ impl From for SessionState { impl From for SessionStructure { fn from(value: SessionState) -> SessionStructure { - value.session + value.into_protobuf() } } impl From<&SessionState> for SessionStructure { fn from(value: &SessionState) -> SessionStructure { - value.session.clone() + value.to_protobuf() } } @@ -1007,7 +1183,7 @@ impl SessionRecord { } let current_session = self .current_session - .map(|state| session_components_from_structure(state.session)) + .map(|state| session_components_from_structure(state.into_protobuf())) .transpose()?; let previous_sessions = self .previous_sessions @@ -1019,7 +1195,7 @@ impl SessionRecord { } let mut state = SessionState::from_session_structure(session); state.fast_forward_sender_chain_or_drop(reserved_sender_chain_index); - session_components_from_structure(state.session) + session_components_from_structure(state.into_protobuf()) }) .collect::, _>>()?; @@ -1076,7 +1252,7 @@ impl SessionRecord { }; let mut state = SessionState::from_session_structure(session); state.fast_forward_sender_chain_or_drop(ceiling); - *archived = ArchivedSession::encode(&state.session); + *archived = ArchivedSession::encode(&state.into_protobuf()); } self.lease.waive(); } @@ -1375,7 +1551,7 @@ impl SessionRecord { sessions.pop(); } current_session.clear_unacknowledged_pre_key_message(); - sessions.insert(0, ArchivedSession::encode(¤t_session.session)); + sessions.insert(0, ArchivedSession::encode(¤t_session.into_protobuf())); true } else { false @@ -1425,10 +1601,12 @@ impl SessionRecord { } let mut cache = buffa::SizeCache::new(); - let current_msg_len = self - .current_session - .as_ref() - .map(|s| s.session.compute_size(&mut cache) as usize); + // Borrowed unless a receiver chain holds skipped keys, in which case + // the backlog is reassembled into a clone for the encoder. + let current = self.current_session.as_ref().map(|s| s.protobuf()); + let current_msg_len = current + .as_deref() + .map(|s| s.compute_size(&mut cache) as usize); let current_len = current_msg_len .map(|msg_len| 1 + varint_len(msg_len as u64) + msg_len) .unwrap_or(0); @@ -1461,10 +1639,10 @@ impl SessionRecord { buf.clear(); buf.reserve(current_len + previous_len + reserved_len + incarnation_len); - if let Some(state) = &self.current_session + if let Some(session) = current.as_deref() && let Some(msg_len) = current_msg_len { - write_len_delimited(1, &state.session, msg_len, &mut cache, buf); + write_len_delimited(1, session, msg_len, &mut cache, buf); } for archived in self.previous_sessions.iter() { let bytes = archived.as_bytes(); @@ -1499,7 +1677,7 @@ impl SessionRecord { let current = self .current_session .as_ref() - .map(|s| session_pointed_bytes(&s.session)) + .map(|s| session_pointed_bytes(&s.session) + s.skipped_pointed_bytes()) .unwrap_or(0); // The `Arc` owns a `Vec` header plus its buffer; each archived state // is its own boxed slice of encoded bytes. @@ -2026,12 +2204,14 @@ mod tests { } } - /// What these reports exist to catch is a receiver chain accumulating - /// skipped message keys — up to `MAX_MESSAGE_KEYS` of them per chain. That - /// only works if the figure is memory, not wire bytes: a modern seed-only - /// key is ~36 bytes encoded and 136 bytes of `MessageKey` plus a 32-byte - /// `Bytes` allocation in memory, because the three fields it leaves empty - /// still occupy their `Option` slots. + /// The skipped-key backlog is the one structure in the store that grows + /// without bound, up to `MAX_MESSAGE_KEYS` per chain, so the report has + /// to charge it at what it costs in memory — and what it costs is now + /// the inline `SkippedKey`, not the 136-byte protobuf `MessageKey` plus a + /// 32-byte `Bytes` allocation that a seed-only key used to occupy. The + /// figure must cover the compact form and must no longer be anywhere + /// near the protobuf one, or the report would be describing memory the + /// state no longer holds. #[test] fn skipped_message_keys_are_reported_at_their_in_memory_cost() { const KEYS: usize = 500; @@ -2057,26 +2237,48 @@ mod tests { let mut session = make_cache_shape_session(1, 0, 0); session.receiver_chains = vec![chain]; + let state = SessionState::from_session_structure(session.clone()); + assert!( + state.session.receiver_chains[0].message_keys.is_empty(), + "the protobuf keeps no skipped keys in memory" + ); let record = SessionRecord { - current_session: Some(SessionState::from_session_structure(session.clone())), + current_session: Some(state), previous_sessions: Arc::new(Vec::new()), lease: CounterLease::default(), }; let reported = record.estimated_size(); - let backlog = KEYS * size_of::(); + let compact = KEYS * size_of::(); + let protobuf_shape = KEYS * (size_of::() + 32); assert!( - reported >= backlog, - "a {KEYS}-key backlog occupies at least {backlog} B; reported {reported} B" + reported >= compact, + "a {KEYS}-key backlog occupies at least {compact} B; reported {reported} B" ); - - // And the figure this replaced: the encoded size, which is what the - // report used to hand back for the very same record. - let encoded = session.encode_to_vec().len(); assert!( - reported > encoded * 3, - "the in-memory figure ({reported} B) must be far above the encoded \ - one ({encoded} B) for a seed-only backlog" + reported < protobuf_shape, + "the report ({reported} B) must not still charge the protobuf shape \ + ({protobuf_shape} B) the backlog no longer takes" + ); + assert!( + size_of::() <= 40, + "a seed-only skipped key is a u32 and 32 bytes of seed" + ); + + // And the backlog round-trips: what was moved out comes back on encode. + assert_eq!( + record.serialize().expect("serialize").len(), + SessionRecord { + current_session: Some(SessionState { + session: session.clone(), + skipped: Vec::new(), + }), + previous_sessions: Arc::new(Vec::new()), + lease: CounterLease::default(), + } + .serialize() + .expect("serialize") + .len() ); } @@ -2580,7 +2782,7 @@ mod tests { let key = KeyPair::generate(&mut rng()).public_key; let state = create_test_session_state(3, &key); Arc::make_mut(&mut record.previous_sessions) - .push(ArchivedSession::encode(&state.session)); + .push(ArchivedSession::encode(&state.to_protobuf())); } // Serialize From 5853a1cab001d2736f512ae36e7eeab716e08d52 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 17:12:36 +0000 Subject: [PATCH 2/4] perf(types)!: shrink MessageInfo and make message ids inline MessageInfo is retained per message through the commit batch and by every consumer that keeps a message; it was 968 bytes, most of it fields that are absent on the ordinary message. `meta_info` (280 B), `bot_info` and `device_sent_meta` are now boxed options, built only when their stanza children are present; `MessageInfo::meta()` hands readers an empty one otherwise. `ephemeral_expiration` and `comment_target` move to `InboundMessage`, where they are known: writing them into the shared `Arc` deep-copied the whole struct on every disappearing-chat message. `MessageId` becomes `CompactString`, so a 22-character id lives inline in the info, in `ChatMessageId`/`SenderMessageId` cache keys and in receipt id lists instead of costing a heap allocation each; `push_name` follows for the same reason. BREAKING: `MessageInfo` field types change as described above and `ephemeral_expiration`/`comment_target` move to `InboundMessage`; `MessageId` is a `CompactString`. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_0172fpxasGTrouFyYH5UGmjN --- examples/durability_hook.rs | 2 +- src/bot.rs | 6 +-- src/client/messaging.rs | 2 +- src/client/tests.rs | 12 ++--- src/features/contacts.rs | 2 +- src/history_sync.rs | 8 ++-- src/message/commit_batch.rs | 4 +- src/message/dispatch.rs | 36 ++++++++------- src/message/durability.rs | 2 +- src/message/msg_secret.rs | 6 +-- src/message/receive.rs | 2 +- src/message/tests.rs | 70 ++++++++++++++--------------- src/pdo.rs | 58 +++++++++++------------- src/receipt.rs | 57 +++++++++++------------ src/retry.rs | 24 +++++----- tests/e2e/tests/offline_receipts.rs | 8 ++-- tests/e2e/tests/receipts.rs | 18 ++++---- wacore/binary/src/jid.rs | 5 ++- wacore/binary/src/node.rs | 7 +++ wacore/src/history_sync.rs | 34 +++++++------- wacore/src/iq/business.rs | 2 +- wacore/src/messages.rs | 59 +++++++++++++----------- wacore/src/stanza/business.rs | 2 +- wacore/src/stanza/receipt.rs | 40 +++++++++-------- wacore/src/types/events.rs | 30 +++++++++---- wacore/src/types/message.rs | 45 ++++++++++++------- 26 files changed, 289 insertions(+), 252 deletions(-) diff --git a/examples/durability_hook.rs b/examples/durability_hook.rs index 85b253243..3f8405e5a 100644 --- a/examples/durability_hook.rs +++ b/examples/durability_hook.rs @@ -118,7 +118,7 @@ impl InboundDurabilityHook for InboxArchiver { let key: CommitKey = ( m.info.source.chat.to_string(), m.info.source.sender.to_string(), - m.info.id.clone(), + m.info.id.to_string(), ); // Dedup against the archive AND earlier entries of this same // batch, so one fsync can never append a key twice. diff --git a/src/bot.rs b/src/bot.rs index 17f0dff08..81bd24cac 100644 --- a/src/bot.rs +++ b/src/bot.rs @@ -167,7 +167,7 @@ impl MessageContext { // info.source.chat, so remote_jid is omitted (WA Web parity). let chat = &self.info.source.chat; wacore::proto_helpers::build_quote_context_with_info( - &self.info.id, + self.info.id.as_str(), &self.info.source.sender, chat, chat, @@ -184,7 +184,7 @@ impl MessageContext { wa::MessageKey { remote_jid: Some(self.info.source.chat.to_string()), from_me: Some(self.info.source.is_from_me), - id: Some(self.info.id.clone()), + id: Some(self.info.id.to_string()), participant: needs_participant.then(|| self.info.source.sender.to_string()), } } @@ -2309,7 +2309,7 @@ mod tests { fn react_info(chat: &str, sender: &str, id: &str, is_group: bool) -> MessageInfo { use crate::types::message::MessageSource; MessageInfo { - id: id.to_string(), + id: id.into(), source: MessageSource { chat: chat.parse().expect("chat jid"), sender: sender.parse().expect("sender jid"), diff --git a/src/client/messaging.rs b/src/client/messaging.rs index 56b4ab5a2..461ca9645 100644 --- a/src/client/messaging.rs +++ b/src/client/messaging.rs @@ -423,7 +423,7 @@ impl Client { ChatMessageId { chat, - id: id.to_owned(), + id: id.into(), } } diff --git a/src/client/tests.rs b/src/client/tests.rs index 873674221..7082f06ad 100644 --- a/src/client/tests.rs +++ b/src/client/tests.rs @@ -2978,7 +2978,7 @@ fn test_message_ack_source_node_own_device_addressing() { // Own-account branch: sender == `from` (device-qualified), chat is the // device-stripped recipient. `to` must come from sender, not chat. let info = MessageInfo { - id: "AC055553E56A2C12DE592DAD6353C477".to_string(), + id: "AC055553E56A2C12DE592DAD6353C477".into(), source: MessageSource { sender: "236395184570386@lid".parse().expect("sender"), chat: "156535032389744@lid".parse().expect("chat"), @@ -3023,7 +3023,7 @@ fn test_message_ack_source_node_own_device_addressing() { fn test_message_ack_source_node_incoming_dm_addressing() { use crate::types::message::{MessageInfo, MessageSource}; let info = MessageInfo { - id: "MSGID".to_string(), + id: "MSGID".into(), source: MessageSource { sender: "5511999998888:3@s.whatsapp.net".parse().expect("sender"), chat: "5511999998888@s.whatsapp.net".parse().expect("chat"), @@ -3057,7 +3057,7 @@ fn test_message_ack_source_node_incoming_dm_addressing() { fn test_message_ack_source_node_status_addressing() { use crate::types::message::{MessageInfo, MessageSource}; let info = MessageInfo { - id: "STATUSMSG".to_string(), + id: "STATUSMSG".into(), source: MessageSource { chat: "status@broadcast".parse().expect("status chat"), sender: "181531758878822@lid".parse().expect("participant"), @@ -3096,7 +3096,7 @@ fn test_message_ack_source_node_group_addressing() { use crate::types::message::{MessageInfo, MessageSource}; // Group branch: chat == group `from`, sender == participant. let info = MessageInfo { - id: "GROUPMSGID".to_string(), + id: "GROUPMSGID".into(), source: MessageSource { chat: "120363011111111111@g.us".parse().expect("group"), sender: "181531758878822@lid".parse().expect("participant"), @@ -4020,7 +4020,7 @@ async fn a_panicking_observer_leaves_the_client_sending() { fn receipt_test_info(id: &str) -> Arc { Arc::new(crate::types::message::MessageInfo { - id: id.to_string(), + id: id.into(), source: crate::types::message::MessageSource { chat: "15550001111@s.whatsapp.net".parse().unwrap(), sender: "15550001111@s.whatsapp.net".parse().unwrap(), @@ -5020,7 +5020,7 @@ async fn memory_report_on_fresh_client() { // Retained bytes must appear once something is cached. let key = ChatMessageId::new( "559980000001@s.whatsapp.net".parse().unwrap(), - "3EB0TESTMSGID".to_string(), + "3EB0TESTMSGID".into(), ); client .recent_messages diff --git a/src/features/contacts.rs b/src/features/contacts.rs index 3645d4d82..54ffd6a77 100644 --- a/src/features/contacts.rs +++ b/src/features/contacts.rs @@ -361,7 +361,7 @@ mod tests { #[test] fn test_profile_picture_struct() { let pic = ProfilePicture { - id: "123456789".to_string(), + id: "123456789".into(), url: "https://example.com/pic.jpg".to_string(), direct_path: Some("/v/pic.jpg".to_string()), hash: None, diff --git a/src/history_sync.rs b/src/history_sync.rs index aa21550b7..5191362d0 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -839,7 +839,7 @@ mod tests { let history_sync = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![wa::HistorySyncMsg { message: buffa::MessageField::some(wa::WebMessageInfo { key: buffa::MessageField::some(wa::MessageKey { @@ -927,7 +927,7 @@ mod tests { let history_sync = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), ..Default::default() }], ..Default::default() @@ -987,7 +987,7 @@ mod tests { let history_sync = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![wa::HistorySyncMsg { message: buffa::MessageField::some(wa::WebMessageInfo { key: buffa::MessageField::some(wa::MessageKey { @@ -1078,7 +1078,7 @@ mod tests { let history_sync = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages, ..Default::default() }], diff --git a/src/message/commit_batch.rs b/src/message/commit_batch.rs index 01f04d3c4..30a7348f4 100644 --- a/src/message/commit_batch.rs +++ b/src/message/commit_batch.rs @@ -1086,7 +1086,7 @@ mod tests { self.batches .lock() .expect("hook lock") - .push(batch.iter().map(|m| m.info.id.clone()).collect()); + .push(batch.iter().map(|m| m.info.id.to_string()).collect()); Ok(()) } } @@ -1098,7 +1098,7 @@ mod tests { ..Default::default() })) .info(Arc::new(MessageInfo { - id: id.to_string(), + id: id.into(), source: MessageSource { chat: "100@g.us".parse().unwrap(), sender: "200@s.whatsapp.net".parse().unwrap(), diff --git a/src/message/dispatch.rs b/src/message/dispatch.rs index 50c777e2a..aeabadb46 100644 --- a/src/message/dispatch.rs +++ b/src/message/dispatch.rs @@ -89,29 +89,27 @@ impl Client { wacore::telemetry::recv("decrypted"); self.stats.record_message_received(); - let mut info = Arc::clone(info); - if info.ephemeral_expiration.is_none() - && let Some(exp) = msg.get_base_message().get_ephemeral_expiration() - { - Arc::make_mut(&mut info).ephemeral_expiration = Some(exp); - } + // Both ride on the `InboundMessage`, not the shared `MessageInfo`: + // writing them into the `Arc` every `` of the stanza holds + // deep-copied the whole info on every disappearing-chat message. + let ephemeral_expiration = msg.get_base_message().get_ephemeral_expiration(); // Keep this ordered with dispatch; add-on messages can immediately // reference the secret from the stanza just processed. - self.maybe_capture_inbound_msg_secret(&msg, &info).await; + self.maybe_capture_inbound_msg_secret(&msg, info).await; let decrypted = self - .maybe_decrypt_secret_encrypted_message(&msg, &info) + .maybe_decrypt_secret_encrypted_message(&msg, info) .await; // A decrypted comment surfaces as its inner body Message, which has no - // slot for the parent post key; carry the threading link on the info. - if decrypted.is_some() - && let Some(target) = msg - .enc_comment_message + // slot for the parent post key; carry the threading link beside it. + let comment_target = if decrypted.is_some() { + msg.enc_comment_message .as_option() .and_then(|c| c.target_message_key.as_option().cloned()) - { - Arc::make_mut(&mut info).comment_target = Some(target); - } + .map(Box::new) + } else { + None + }; let dispatch_msg = Arc::new(decrypted.unwrap_or(msg)); // Newsletters never enter the commit pipeline: the plaintext stanza @@ -126,7 +124,9 @@ impl Client { .messages(Arc::from([wacore::types::events::InboundMessage::builder( ) .message(dispatch_msg) - .info(info) + .info(Arc::clone(info)) + .maybe_ephemeral_expiration(ephemeral_expiration) + .maybe_comment_target(comment_target) .build()])) .origin(wacore::types::events::BatchOrigin::Live) .build(), @@ -141,7 +141,9 @@ impl Client { self.commit_or_batch_inbound( wacore::types::events::InboundMessage::builder() .message(dispatch_msg) - .info(info) + .info(Arc::clone(info)) + .maybe_ephemeral_expiration(ephemeral_expiration) + .maybe_comment_target(comment_target) .build(), track_commit, ) diff --git a/src/message/durability.rs b/src/message/durability.rs index 167d00e40..1d791753a 100644 --- a/src/message/durability.rs +++ b/src/message/durability.rs @@ -133,7 +133,7 @@ mod tests { fn test_info(id: &str) -> Arc { use crate::types::message::MessageSource; Arc::new(MessageInfo { - id: id.to_string(), + id: id.into(), source: MessageSource { chat: "100@g.us".parse().unwrap(), sender: "200@s.whatsapp.net".parse().unwrap(), diff --git a/src/message/msg_secret.rs b/src/message/msg_secret.rs index 048bbfd9a..b541c5ac8 100644 --- a/src/message/msg_secret.rs +++ b/src/message/msg_secret.rs @@ -558,7 +558,7 @@ impl Client { // Chat scope for the secret lookup: prefer ; // fall back to the stanza's chat (matches WA Web `decryptMsmsgBotMessage`). let chat_for_lookup = info - .meta_info + .meta() .target_chat .as_ref() .unwrap_or(&info.source.chat) @@ -569,7 +569,7 @@ impl Client { // The id used for the SECRET LOOKUP is `meta.target_id` (our outbound // id); the id used as HKDF input is the bot reply id (or // `bot_info.edit_target_id` when the bot is editing a prior reply). - let target_id = match info.meta_info.target_id.as_deref() { + let target_id = match info.meta().target_id.as_deref() { Some(id) => id, None => { log::warn!( @@ -839,7 +839,7 @@ impl Client { /// Resolve `target_sender` for a msmsg stanza: echo from `` when /// present, else fall back to our LID (sender on bot server) or PN. async fn resolve_msmsg_target_sender(&self, info: &Arc) -> Option { - if let Some(ts) = info.meta_info.target_sender.as_ref() { + if let Some(ts) = info.meta().target_sender.as_ref() { return Some(ts.clone()); } if info.source.sender.server == wacore_binary::Server::Bot { diff --git a/src/message/receive.rs b/src/message/receive.rs index 08d0622ab..e38c6ae9e 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -1895,7 +1895,7 @@ impl Client { // `WAWebHandleHistorySyncNotification` gates on `isMePrimaryNonLid`. if let Some(history_sync) = history_sync_taken { if info.source.is_from_me { - self.handle_history_sync(info.id.clone(), history_sync) + self.handle_history_sync(info.id.to_string(), history_sync) .await; } else { warn!( diff --git a/src/message/tests.rs b/src/message/tests.rs index f85b9c806..20cbf12c6 100644 --- a/src/message/tests.rs +++ b/src/message/tests.rs @@ -379,7 +379,7 @@ async fn batch_accumulates_undecryptable_and_dispatches_once() { .parse() .expect("test JID should be valid"); let info = Arc::new(MessageInfo { - id: "BATCH_UNDEC_ONCE".to_string(), + id: "BATCH_UNDEC_ONCE".into(), source: crate::types::message::MessageSource { sender: sender_jid.clone(), chat: sender_jid.clone(), @@ -973,7 +973,7 @@ async fn migration_plaintext_failure_nacks_without_signal_retry() { let bad_old_pn_msg = alice_old.encrypt(&bob_addr, &[0xff, 0x01]).await; let payloads = vec![enc_payload_from_ciphertext(&bad_old_pn_msg)]; let info = Arc::new(MessageInfo { - id: "MIGRATION_BAD_PLAINTEXT".to_string(), + id: "MIGRATION_BAD_PLAINTEXT".into(), source: crate::types::message::MessageSource { sender: alice_lid.clone(), chat: alice_lid.clone(), @@ -1102,7 +1102,7 @@ async fn test_badmac_preserves_session() { let enc_ref = enc_node.as_node_ref(); let payloads: Vec = vec![EncPayload::from_node_ref(&enc_ref, 0).unwrap()]; let info = Arc::new(MessageInfo { - id: "BADMAC_TAMPER_MSG".to_string(), + id: "BADMAC_TAMPER_MSG".into(), source: crate::types::message::MessageSource { sender: alice.jid.clone(), chat: alice.jid.clone(), @@ -1262,7 +1262,7 @@ async fn test_prod_scenario_pkmsg_archives_old_session_after_badmac() { let enc_ref = enc_node.as_node_ref(); let payloads: Vec = vec![EncPayload::from_node_ref(&enc_ref, 0).unwrap()]; let info = Arc::new(MessageInfo { - id: "PROD_LOOP_REPRO_STALE".to_string(), + id: "PROD_LOOP_REPRO_STALE".into(), source: crate::types::message::MessageSource { sender: alice.jid.clone(), chat: alice.jid.clone(), @@ -3326,13 +3326,13 @@ async fn test_pn_message_uses_pn_when_no_lid_mapping() { /// Helper to create a test MessageInfo with customizable fields fn create_test_message_info(chat: &str, msg_id: &str, sender: &str) -> MessageInfo { - use wacore::types::message::{EditAttribute, MessageCategory, MessageSource, MsgMetaInfo}; + use wacore::types::message::{EditAttribute, MessageCategory, MessageSource}; let chat_jid: Jid = chat.parse().expect("valid chat JID"); let sender_jid: Jid = sender.parse().expect("valid sender JID"); MessageInfo { - id: msg_id.to_string(), + id: msg_id.into(), server_id: 0, r#type: Some(wacore::types::message::StanzaMessageType::Text), source: MessageSource { @@ -3347,23 +3347,21 @@ fn create_test_message_info(chat: &str, msg_id: &str, sender: &str) -> MessageIn recipient: None, }, timestamp: wacore::time::now_utc(), - push_name: "Test User".to_string(), + push_name: "Test User".into(), category: MessageCategory::default(), multicast: false, media_type: None, edit: EditAttribute::default(), bot_info: None, - meta_info: MsgMetaInfo::default(), + meta_info: None, verified_name: None, device_sent_meta: None, - ephemeral_expiration: None, is_offline: false, unavailable_request_id: None, server_timestamp_us: None, verified_level: None, verified_name_serial: None, peer_recipient_pn: None, - comment_target: None, bcl_participants: Vec::new(), } } @@ -5132,7 +5130,7 @@ async fn undecryptable_receive_branch_stays_silent_when_batch_dispatched() { .expect("test JID should be valid"); let msg_id = "UNDEC_LOG_BATCH_OWNS_DISPATCH"; let info = Arc::new(MessageInfo { - id: msg_id.to_string(), + id: msg_id.into(), source: crate::types::message::MessageSource { sender: sender.clone(), chat: sender.clone(), @@ -5222,7 +5220,7 @@ async fn undecryptable_receive_branch_announces_the_dispatch_it_performs() { .expect("test JID should be valid"); let msg_id = "UNDEC_LOG_BRANCH_DISPATCHES"; let info = Arc::new(MessageInfo { - id: msg_id.to_string(), + id: msg_id.into(), source: crate::types::message::MessageSource { sender: participant, chat: group.clone(), @@ -7216,7 +7214,7 @@ async fn decrypt_failure_emits_transport_ack() { let sender: Jid = "236395184570386@lid".parse().expect("sender JID"); let recipient: Jid = "156535032389744@lid".parse().expect("recipient JID"); let info = Arc::new(MessageInfo { - id: "AC055553E56A2C12DE592DAD6353C477".to_string(), + id: "AC055553E56A2C12DE592DAD6353C477".into(), source: crate::types::message::MessageSource { sender: sender.clone(), chat: recipient.clone(), @@ -7264,7 +7262,7 @@ async fn decrypt_failure_emits_transport_ack() { async fn self_fanout_decrypt_failure_acked_via_sender_receipt() { let (client, transport) = capturing_client("self_fanout_badmac").await; let info = Arc::new(MessageInfo { - id: "AC00000000000000000000000000BEEF".to_string(), + id: "AC00000000000000000000000000BEEF".into(), source: crate::types::message::MessageSource { sender: "100000000000001@lid".parse().expect("sender"), chat: "200000000000002@bot".parse().expect("chat"), @@ -7326,7 +7324,7 @@ async fn self_fanout_decrypt_failure_acked_via_sender_receipt() { async fn bot_author_self_fanout_decrypt_failure_not_sender_receipt() { let (client, transport) = capturing_client("bot_author_badmac").await; let info = Arc::new(MessageInfo { - id: "OWNBOTFAIL1".to_string(), + id: "OWNBOTFAIL1".into(), source: crate::types::message::MessageSource { sender: "100000000000002@bot".parse().expect("sender"), chat: "300000000000003@lid".parse().expect("chat"), @@ -7376,7 +7374,7 @@ async fn decrypt_failure_does_not_ack_when_retry_send_fails() { let (client, transport) = capturing_client("retry_fail_no_ack").await; let sender: Jid = "5511777776666@s.whatsapp.net".parse().expect("sender"); let info = Arc::new(MessageInfo { - id: "NOACK1".to_string(), + id: "NOACK1".into(), source: crate::types::message::MessageSource { sender: sender.clone(), chat: sender.clone(), @@ -7409,7 +7407,7 @@ async fn decrypt_failure_sends_retry_before_ack() { let (client, transport) = capturing_client("retry_before_ack").await; let sender: Jid = "5511777776666@s.whatsapp.net".parse().expect("sender"); let info = Arc::new(MessageInfo { - id: "RBA1".to_string(), + id: "RBA1".into(), source: crate::types::message::MessageSource { sender: sender.clone(), chat: sender.clone(), @@ -7466,7 +7464,7 @@ async fn decrypt_failure_sends_retry_before_ack() { async fn status_broadcast_decrypt_failure_acks_to_chat() { let (client, transport) = capturing_client("status_fail_ack").await; let info = Arc::new(MessageInfo { - id: "STATUSMSGID".to_string(), + id: "STATUSMSGID".into(), source: crate::types::message::MessageSource { sender: "236395184570386@lid".parse().expect("sender"), chat: "status@broadcast".parse().expect("status chat"), @@ -7517,7 +7515,7 @@ async fn process_session_ct( let enc_ref = enc.as_node_ref(); let payload = EncPayload::from_node_ref(&enc_ref, 0).unwrap(); let info = Arc::new(MessageInfo { - id: id.to_string(), + id: id.into(), source: crate::types::message::MessageSource { sender: sender.clone(), chat: sender.clone(), @@ -7573,7 +7571,7 @@ fn msmsg_payload_from_bytes(bytes: Vec) -> EncPayload { fn group_message_info(id: &str, group: &Jid, sender: &Jid, is_from_me: bool) -> Arc { Arc::new(MessageInfo { - id: id.to_string(), + id: id.into(), source: crate::types::message::MessageSource { sender: sender.clone(), chat: group.clone(), @@ -8354,7 +8352,7 @@ async fn error_message_ack_is_not_counted_as_positive_confirmation() { let (client, transport) = capturing_client("error_ack_not_positive").await; let id = "ERROR_ACK_NOT_POSITIVE"; let info = Arc::new(MessageInfo { - id: id.to_string(), + id: id.into(), source: crate::types::message::MessageSource { sender: "146824178450530@lid".parse().expect("sender"), chat: "120363408782575448@g.us".parse().expect("group"), @@ -8566,7 +8564,7 @@ async fn duplicate_message_is_acked_with_delivery_receipt() { async fn own_self_fanout_acked_via_sender_receipt() { let (client, transport) = capturing_client("own_ack").await; let own = Arc::new(MessageInfo { - id: "OWN1".to_string(), + id: "OWN1".into(), source: crate::types::message::MessageSource { sender: "100000000000001@lid".parse().expect("sender"), chat: "300000000000003@lid".parse().expect("chat"), @@ -8617,7 +8615,7 @@ async fn own_self_fanout_acked_via_sender_receipt() { async fn bot_self_fanout_acked_via_sender_receipt() { let (client, transport) = capturing_client("bot_self_fanout").await; let own = Arc::new(MessageInfo { - id: "AC00000000000000000000000000BEEF".to_string(), + id: "AC00000000000000000000000000BEEF".into(), source: crate::types::message::MessageSource { // from = our own LID with its device (the server fans our // outgoing bot prompt back to this device); chat = the bot @@ -8669,7 +8667,7 @@ async fn bot_self_fanout_acked_via_sender_receipt() { async fn own_bot_author_dm_acks_not_sender_receipt() { let (client, transport) = capturing_client("own_bot_author").await; let own = Arc::new(MessageInfo { - id: "OWNBOT1".to_string(), + id: "OWNBOT1".into(), source: crate::types::message::MessageSource { sender: "100000000000002@bot".parse().expect("sender"), chat: "300000000000003@lid".parse().expect("chat"), @@ -12553,7 +12551,7 @@ async fn enc_reaction_inbound_decrypts_to_plaintext_shape() { ..Default::default() }; let info = Arc::new(MessageInfo { - id: "REACT1".to_string(), + id: "REACT1".into(), source: MessageSource { chat: group.clone(), sender: reactor.clone(), @@ -12659,7 +12657,7 @@ async fn enc_comment_inbound_dispatches_body_with_parent_link() { ..Default::default() }; let info = Arc::new(MessageInfo { - id: COMMENT_ID.to_string(), + id: COMMENT_ID.into(), source: MessageSource { chat: group.clone(), sender: commenter.clone(), @@ -12679,7 +12677,7 @@ async fn enc_comment_inbound_dispatches_body_with_parent_link() { 'outer: while tokio::time::Instant::now() < deadline { while let Ok(event) = rx.try_recv() { if let Some(m) = event.messages().find(|m| m.info.id == COMMENT_ID) { - let (msg, info) = (&m.message, &m.info); + let msg = &m.message; seen = true; assert_eq!( msg.extended_text_message @@ -12693,7 +12691,7 @@ async fn enc_comment_inbound_dispatches_body_with_parent_link() { "the envelope must not survive substitution" ); assert_eq!( - info.comment_target.as_ref().and_then(|k| k.id.as_deref()), + m.comment_target.as_ref().and_then(|k| k.id.as_deref()), Some(PARENT_ID), "the parent post key must surface on the info" ); @@ -12766,7 +12764,7 @@ async fn addon_decrypts_right_after_capture_without_flush() { }; let mk_info = |id: &str| { Arc::new(MessageInfo { - id: id.to_string(), + id: id.into(), source: MessageSource { chat: chat.parse().expect("chat"), sender: chat.parse().expect("sender"), @@ -13025,7 +13023,7 @@ async fn test_invalid_signed_prekey_id_sends_retry_receipt() { let enc_ref = enc_node.as_node_ref(); let payloads: Vec = vec![EncPayload::from_node_ref(&enc_ref, 0).unwrap()]; let info = Arc::new(MessageInfo { - id: "INVALID_SPK_ID_MSG".to_string(), + id: "INVALID_SPK_ID_MSG".into(), source: crate::types::message::MessageSource { sender: alice.jid.clone(), chat: alice.jid.clone(), @@ -13589,7 +13587,7 @@ fn enc_payload_at(enc_type: &str, bytes: Vec, enc_index: usize) -> EncPayloa fn dm_info(msg_id: &str, sender: &Jid) -> Arc { Arc::new(MessageInfo { - id: msg_id.to_string(), + id: msg_id.into(), source: crate::types::message::MessageSource { sender: sender.clone(), chat: sender.clone(), @@ -13940,7 +13938,7 @@ async fn a_group_enc_without_a_sender_key_reports_no_sender_key() { let group: Jid = "120363000000000002@g.us".parse().unwrap(); let participant: Jid = "5511900000003:1@s.whatsapp.net".parse().unwrap(); let info = Arc::new(MessageInfo { - id: "ENCFAIL_NOSK".to_string(), + id: "ENCFAIL_NOSK".into(), source: crate::types::message::MessageSource { sender: participant, chat: group.clone(), @@ -13993,7 +13991,7 @@ async fn a_skmsg_skipped_after_a_session_failure_reports_not_attempted() { let group: Jid = "120363000000000003@g.us".parse().unwrap(); let participant: Jid = "5511900000004@s.whatsapp.net".parse().unwrap(); let info = Arc::new(MessageInfo { - id: "ENCFAIL_SKIPPED".to_string(), + id: "ENCFAIL_SKIPPED".into(), source: crate::types::message::MessageSource { sender: participant.clone(), chat: group, @@ -14049,7 +14047,7 @@ async fn session_encs_from_a_group_sender_report_not_attempted() { let group: Jid = "120363000000000004@g.us".parse().unwrap(); let participant: Jid = "5511900000005@s.whatsapp.net".parse().unwrap(); let info = Arc::new(MessageInfo { - id: "ENCFAIL_GROUPSENDER".to_string(), + id: "ENCFAIL_GROUPSENDER".into(), source: crate::types::message::MessageSource { sender: participant, chat: group.clone(), @@ -14331,7 +14329,7 @@ async fn a_group_envelope_that_does_not_parse_reports_malformed_ciphertext() { let group: Jid = "120363000000000005@g.us".parse().unwrap(); let participant: Jid = "5511900000009@s.whatsapp.net".parse().unwrap(); let info = Arc::new(MessageInfo { - id: "ENCFAIL_SKMSG_PARSE".to_string(), + id: "ENCFAIL_SKMSG_PARSE".into(), source: crate::types::message::MessageSource { sender: participant, chat: group.clone(), @@ -14587,7 +14585,7 @@ async fn a_skmsg_skipped_beside_a_duplicate_is_still_reported() { // to skip the group payload. let group: Jid = "120363000000000006@g.us".parse().unwrap(); let redelivered = Arc::new(MessageInfo { - id: "ENCFAIL_DUP_REDELIVERED".to_string(), + id: "ENCFAIL_DUP_REDELIVERED".into(), source: crate::types::message::MessageSource { sender: alice.jid.clone(), chat: group, diff --git a/src/pdo.rs b/src/pdo.rs index 8e8085283..6c4533aa5 100644 --- a/src/pdo.rs +++ b/src/pdo.rs @@ -18,9 +18,7 @@ use crate::client::Client; use crate::types::message::MessageInfo; use log::{debug, info, warn}; use std::sync::Arc; -use wacore::types::message::{ - ChatMessageId, EditAttribute, MessageCategory, MessageSource, MsgMetaInfo, -}; +use wacore::types::message::{ChatMessageId, EditAttribute, MessageCategory, MessageSource}; use wacore_binary::{Jid, JidExt}; use waproto::whatsapp as wa; @@ -179,7 +177,7 @@ impl Client { let message_key = wa::MessageKey { remote_jid: Some(resolved_jid.to_string()), from_me: Some(info.source.is_from_me), - id: Some(info.id.clone()), + id: Some(info.id.to_string()), participant: participant.map(|p| p.to_string()), }; @@ -815,7 +813,7 @@ impl Client { let response_from_me = key.from_me.unwrap_or(false); let cache_key = match remote_jid_str.parse::() { - Ok(jid) => ChatMessageId::new(jid, msg_id.to_owned()), + Ok(jid) => ChatMessageId::new(jid, msg_id.into()), Err(_) => { warn!( "PDO response has unparseable remote_jid: {}", @@ -898,18 +896,15 @@ impl Client { return; }; - { + let ephemeral_expiration = { use wacore::proto_helpers::MessageExt; - let mi = Arc::make_mut(&mut message_info); - if mi.ephemeral_expiration.is_none() { - mi.ephemeral_expiration = message.get_base_message().get_ephemeral_expiration(); - } - mi.unavailable_request_id = if request_id.is_empty() { - None - } else { - Some(request_id.to_owned()) - }; - } + message.get_base_message().get_ephemeral_expiration() + }; + Arc::make_mut(&mut message_info).unavailable_request_id = if request_id.is_empty() { + None + } else { + Some(request_id.to_owned()) + }; info!( "Dispatching PDO-recovered message {} from {} via phone (request_id={})", @@ -928,6 +923,7 @@ impl Client { ) .message(Arc::from(message)) .info(message_info) + .maybe_ephemeral_expiration(ephemeral_expiration) .build()])) .origin(wacore::types::events::BatchOrigin::Live) .build(), @@ -994,7 +990,7 @@ impl Client { .unwrap_or_else(wacore::time::now_utc); Ok(MessageInfo { - id: id.unwrap_or_default().to_owned(), + id: id.unwrap_or_default().into(), server_id: 0, r#type: None, source: MessageSource { @@ -1009,23 +1005,21 @@ impl Client { recipient: None, }, timestamp, - push_name: push_name.unwrap_or_default().to_owned(), + push_name: push_name.unwrap_or_default().into(), category: MessageCategory::default(), multicast: false, media_type: None, edit: EditAttribute::default(), bot_info: None, - meta_info: MsgMetaInfo::default(), + meta_info: None, verified_name: None, device_sent_meta: None, - ephemeral_expiration: None, is_offline: false, unavailable_request_id: None, server_timestamp_us: None, verified_level: None, verified_name_serial: None, peer_recipient_pn: None, - comment_target: None, bcl_participants: Vec::new(), }) } @@ -1292,7 +1286,7 @@ mod tests { ) -> std::sync::Arc { use wacore::types::message::{MessageInfo, MessageSource}; std::sync::Arc::new(MessageInfo { - id: id.to_owned(), + id: id.into(), source: MessageSource { chat: chat.parse().expect("chat jid"), sender: sender.parse().expect("sender jid"), @@ -1488,7 +1482,7 @@ mod tests { let chat = "120363000000000001@g.us"; let msg_id = "PDO_ATTRIBUTION"; - let key = ChatMessageId::new(chat.parse().expect("chat jid"), msg_id.to_owned()); + let key = ChatMessageId::new(chat.parse().expect("chat jid"), msg_id.into()); // Whoever holds the slot when the response lands is not who it answers. client @@ -1506,7 +1500,7 @@ mod tests { key: buffa::MessageField::some(waproto::whatsapp::MessageKey { remote_jid: Some(chat.to_owned()), from_me: Some(false), - id: Some(msg_id.to_owned()), + id: Some(msg_id.into()), participant: Some("111222333444555@lid".to_owned()), }), message: buffa::MessageField::some(waproto::whatsapp::Message { @@ -1564,7 +1558,7 @@ mod tests { let chat = "120363000000000001@g.us"; let msg_id = "PDO_LID_ALIAS"; - let key = ChatMessageId::new(chat.parse().expect("chat jid"), msg_id.to_owned()); + let key = ChatMessageId::new(chat.parse().expect("chat jid"), msg_id.into()); // What a LID-addressed group delivery leaves behind: LID in `sender`, // the PN the stanza carried in `sender_alt`. @@ -1586,7 +1580,7 @@ mod tests { key: buffa::MessageField::some(waproto::whatsapp::MessageKey { remote_jid: Some(chat.to_owned()), from_me: Some(false), - id: Some(msg_id.to_owned()), + id: Some(msg_id.into()), // The phone answers in PN. participant: Some("15550001234@s.whatsapp.net".to_owned()), }), @@ -1644,11 +1638,11 @@ mod tests { let peer = "5511999998888@s.whatsapp.net"; let msg_id = "PDO_DIRECTION"; - let key = ChatMessageId::new(peer.parse().expect("chat jid"), msg_id.to_owned()); + let key = ChatMessageId::new(peer.parse().expect("chat jid"), msg_id.into()); // The slot holds the outgoing half of the conversation. let outgoing = MessageInfo { - id: msg_id.to_owned(), + id: msg_id.into(), source: MessageSource { chat: peer.parse().expect("chat jid"), sender: peer.parse().expect("chat jid"), @@ -1673,7 +1667,7 @@ mod tests { key: buffa::MessageField::some(waproto::whatsapp::MessageKey { remote_jid: Some(peer.to_owned()), from_me: Some(false), - id: Some(msg_id.to_owned()), + id: Some(msg_id.into()), participant: None, }), message: buffa::MessageField::some(waproto::whatsapp::Message { @@ -1762,11 +1756,11 @@ mod tests { let client = setup_reconstruct_client().await; let chat = "5511999998888@s.whatsapp.net"; let msg_id = "PDO_ONCE_3"; - let key = ChatMessageId::new(chat.parse().expect("chat jid"), msg_id.to_owned()); + let key = ChatMessageId::new(chat.parse().expect("chat jid"), msg_id.into()); // A DM: the chat jid is the sender, which is what the gate names. let gate_key = wacore::types::message::SenderMessageId::new( chat.parse().expect("chat jid"), - msg_id.to_owned(), + msg_id.into(), chat.parse().expect("chat jid"), ); @@ -1786,7 +1780,7 @@ mod tests { key: buffa::MessageField::some(waproto::whatsapp::MessageKey { remote_jid: Some(chat.to_owned()), from_me: Some(false), - id: Some(msg_id.to_owned()), + id: Some(msg_id.into()), participant: None, }), ..Default::default() diff --git a/src/receipt.rs b/src/receipt.rs index 2056daf2f..7e47dd1d7 100644 --- a/src/receipt.rs +++ b/src/receipt.rs @@ -568,7 +568,7 @@ impl Client { let mut attrs = nr.attrs(); let from = attrs.jid("from"); let stanza_id = match attrs.optional_string("id") { - Some(id) => id.to_string(), + Some(id) => wacore_binary::MessageId::from(id.as_ref()), None => { log::warn!("Receipt stanza missing required 'id' attribute"); return; @@ -628,10 +628,10 @@ impl Client { wacore::stanza::receipt::parse_participants(part_node); // The event's `message_ids` are `String`, so the borrowed compact id // is widened once here instead of cloning both candidates first. - let fan_out_id: String = agg_msg_id + let fan_out_id: wacore_binary::MessageId = agg_msg_id .as_deref() .or(agg_key.as_deref()) - .map(String::from) + .map(wacore_binary::MessageId::from) .unwrap_or_else(|| stanza_id.clone()); debug!( "Aggregated receipt from {}: stanza={stanza_id} \ @@ -1144,7 +1144,7 @@ mod tests { fn info_with(chat: &str, sender: &str, is_group: bool) -> MessageInfo { MessageInfo { - id: "MID".to_string(), + id: "MID".into(), source: MessageSource { chat: chat.parse().expect("test chat JID"), sender: sender.parse().expect("test sender JID"), @@ -1192,7 +1192,7 @@ mod tests { // recipient=@bot>, `to` preserving the sender's device. Mirrors WA Web // DeliveryReceiptJob (SENDER + USER_JID(recipient)) and whatsmeow. let info = MessageInfo { - id: "FANOUT_BOT".to_string(), + id: "FANOUT_BOT".into(), source: MessageSource { sender: "100000000000001:11@lid".parse().expect("sender"), chat: "200000000000002@bot".parse().expect("chat"), @@ -1227,7 +1227,7 @@ mod tests { // WA Web's `USER_JID` strips the device from `recipient`; a fanout to a // multi-device user echoes the non-AD recipient. let info = MessageInfo { - id: "FANOUT_DEV".to_string(), + id: "FANOUT_DEV".into(), source: MessageSource { sender: "100000000000001:5@lid".parse().expect("sender"), chat: "300000000000003@lid".parse().expect("chat"), @@ -1252,7 +1252,7 @@ mod tests { // recipient) must keep `type="peer_msg"` and carry NO recipient (WA Web // `!l` guard), never `type="sender"`. let info = MessageInfo { - id: "PEER_FANOUT".to_string(), + id: "PEER_FANOUT".into(), source: MessageSource { sender: "100000000000001@lid".parse().expect("sender"), chat: "300000000000003@lid".parse().expect("chat"), @@ -1280,7 +1280,7 @@ mod tests { // type=sender takes precedence over the inactive (passive companion) // branch: a self-fanout is always acknowledged as sender. let info = MessageInfo { - id: "FANOUT_INACTIVE".to_string(), + id: "FANOUT_INACTIVE".into(), source: MessageSource { sender: "100000000000001@lid".parse().expect("sender"), chat: "200000000000002@bot".parse().expect("chat"), @@ -1363,7 +1363,7 @@ mod tests { #[test] fn delivery_receipt_for_lid_dm_preserves_device_in_to() { let info = MessageInfo { - id: "LID_DEV_RECEIPT".to_string(), + id: "LID_DEV_RECEIPT".into(), source: MessageSource { // chat is the non-AD form (matches parse_message_info's // chat = from.to_non_ad()). @@ -1390,7 +1390,7 @@ mod tests { #[test] fn delivery_receipt_for_lid_dm_no_device_unchanged() { let info = MessageInfo { - id: "LID_NO_DEV".to_string(), + id: "LID_NO_DEV".into(), source: MessageSource { chat: "185323896221943@lid".parse().expect("chat"), sender: "185323896221943@lid".parse().expect("sender"), @@ -1411,7 +1411,7 @@ mod tests { #[test] fn delivery_receipt_for_group_to_is_group_not_sender() { let info = MessageInfo { - id: "GRP_RECEIPT".to_string(), + id: "GRP_RECEIPT".into(), source: MessageSource { chat: "120363021033254949@g.us".parse().expect("group"), sender: "156535032389744:7@lid".parse().expect("sender"), @@ -1436,7 +1436,7 @@ mod tests { #[test] fn delivery_receipt_for_peer_dm_to_preserves_device() { let mut info = MessageInfo { - id: "PEER_DEV".to_string(), + id: "PEER_DEV".into(), source: MessageSource { chat: "9999999999@lid".parse().expect("chat"), sender: "9999999999:3@lid".parse().expect("sender"), @@ -1464,7 +1464,7 @@ mod tests { #[test] fn delivery_receipt_for_status_to_is_status_not_sender() { let info = MessageInfo { - id: "STATUS_RECEIPT".to_string(), + id: "STATUS_RECEIPT".into(), source: MessageSource { chat: "status@broadcast".parse().expect("status"), sender: "156535032389744:7@lid".parse().expect("sender"), @@ -1884,7 +1884,7 @@ mod tests { #[test] fn should_send_delivery_receipt_skips_empty_id() { let mut info = info_with("12345@s.whatsapp.net", "12345@s.whatsapp.net", false); - info.id = String::new(); + info.id = Default::default(); assert!(!Client::should_send_delivery_receipt(&info)); } @@ -1963,7 +1963,7 @@ mod tests { .await; let info = MessageInfo { - id: "TEST-ID-123".to_string(), + id: "TEST-ID-123".into(), source: MessageSource { chat: "12345@s.whatsapp.net" .parse() @@ -2007,7 +2007,7 @@ mod tests { .await; let info = MessageInfo { - id: "GROUP-MSG-ID".to_string(), + id: "GROUP-MSG-ID".into(), source: MessageSource { chat: "120363021033254949@g.us" .parse() @@ -2044,7 +2044,7 @@ mod tests { .await; let info = MessageInfo { - id: "OWN-MSG-ID".to_string(), + id: "OWN-MSG-ID".into(), source: MessageSource { chat: "12345@s.whatsapp.net" .parse() @@ -2083,7 +2083,7 @@ mod tests { .await; let info = MessageInfo { - id: "".to_string(), // Empty ID + id: "".into(), // Empty ID source: MessageSource { chat: "12345@s.whatsapp.net" .parse() @@ -2120,7 +2120,7 @@ mod tests { .await; let info = MessageInfo { - id: "STATUS-MSG-ID".to_string(), + id: "STATUS-MSG-ID".into(), source: MessageSource { chat: "status@broadcast" .parse() @@ -2142,7 +2142,7 @@ mod tests { #[test] fn test_should_skip_delivery_receipt_for_newsletter() { let info = MessageInfo { - id: "NEWSLETTER-MSG-ID".to_string(), + id: "NEWSLETTER-MSG-ID".into(), source: MessageSource { chat: "120363173003902460@newsletter" .parse() @@ -2168,7 +2168,7 @@ mod tests { // Self-synced messages (category="peer") should get delivery receipts // even though is_from_me is true. WA Web sends type="peer_msg" for these. let info = MessageInfo { - id: "PEER-MSG-ID".to_string(), + id: "PEER-MSG-ID".into(), source: MessageSource { chat: "155500012345@s.whatsapp.net" .parse() @@ -2725,7 +2725,7 @@ mod tests { fn test_should_skip_non_peer_self_messages() { // Normal self messages (no category) should still be skipped. let info = MessageInfo { - id: "SELF-MSG-ID".to_string(), + id: "SELF-MSG-ID".into(), source: MessageSource { chat: "155500012345@s.whatsapp.net" .parse() @@ -3280,7 +3280,7 @@ mod tests { fn offline_info(id: &str, chat: &str, sender: &str, is_group: bool) -> Arc { let mut info = info_with(chat, sender, is_group); - info.id = id.to_string(); + info.id = id.into(); info.is_offline = true; Arc::new(info) } @@ -3293,7 +3293,7 @@ mod tests { "5511999990000@s.whatsapp.net", false, ); - peer.id = "M6".to_string(); + peer.id = "M6".into(); peer.source.is_from_me = true; peer.category = MessageCategory::Peer; @@ -3390,11 +3390,8 @@ mod tests { // The shape must round-trip through our own ingest parser (the same // form WA Web sends us): list items first, stanza id appended last. let owned = node_to_arc(node.clone()); - let parsed = wacore::stanza::receipt::collect_simple_message_ids( - owned.get(), - "M1".to_string(), - false, - ); + let parsed = + wacore::stanza::receipt::collect_simple_message_ids(owned.get(), "M1".into(), false); assert_eq!( parsed, vec!["M2".to_string(), "M3".to_string(), "M1".to_string()] @@ -3550,7 +3547,7 @@ mod tests { "5511999990000@s.whatsapp.net", false, ); - live.id = "LIVE1".to_string(); + live.id = "LIVE1".into(); client.ack_received_message(&Arc::new(live)); assert_eq!( client.offline_receipt_buffer.lock().expect("buffer").len(), diff --git a/src/retry.rs b/src/retry.rs index a035fb473..91d29e84a 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -2123,7 +2123,7 @@ mod tests { // Case 1: Device sync DM let recipient_lid = Jid::lid("200000000000002"); let device_sync_info = MessageInfo { - id: "DEVICE_SYNC_MSG_001".to_string(), + id: "DEVICE_SYNC_MSG_001".into(), source: MessageSource { chat: recipient_lid.clone(), sender: our_lid.clone(), @@ -2156,7 +2156,7 @@ mod tests { // Case 2: Peer DM with category="peer" let other_pn = Jid::pn("551188888888"); let peer_info = MessageInfo { - id: "PEER123".to_string(), + id: "PEER123".into(), source: MessageSource { chat: other_pn.clone(), sender: our_pn.clone(), @@ -2182,7 +2182,7 @@ mod tests { // Case 3: Group message from our own account let group_info = MessageInfo { - id: "GROUP123".to_string(), + id: "GROUP123".into(), source: MessageSource { chat: "123456789@g.us".parse().unwrap(), sender: our_lid.clone(), @@ -2211,7 +2211,7 @@ mod tests { // Case 4: DM from someone else let other_dm_info = MessageInfo { - id: "OTHER123".to_string(), + id: "OTHER123".into(), source: MessageSource { chat: other_pn.clone(), sender: other_pn.clone(), @@ -3014,7 +3014,7 @@ mod tests { is_group: true, ..Default::default() }) - .message_ids(vec![msg_id.to_string()]) + .message_ids(vec![msg_id.into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(false) @@ -3085,7 +3085,7 @@ mod tests { is_group: chat.is_group(), ..Default::default() }) - .message_ids(vec![msg_id.to_string()]) + .message_ids(vec![msg_id.into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(offline) @@ -3372,7 +3372,7 @@ mod tests { sender: peer, ..Default::default() }) - .message_ids(vec!["DMMISS001".to_string()]) + .message_ids(vec!["DMMISS001".into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(false) @@ -4057,7 +4057,7 @@ mod tests { is_group: true, ..Default::default() }) - .message_ids(vec![msg_id.to_string()]) + .message_ids(vec![msg_id.into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(false) @@ -4575,7 +4575,7 @@ mod tests { sender: from.parse().unwrap(), ..Default::default() }) - .message_ids(vec!["MSG001".to_string()]) + .message_ids(vec!["MSG001".into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(false) @@ -4691,7 +4691,7 @@ mod tests { sender: "236395184570386:33@lid".parse().unwrap(), ..Default::default() }) - .message_ids(vec!["MSG001".to_string()]) + .message_ids(vec!["MSG001".into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(false) @@ -4717,7 +4717,7 @@ mod tests { sender: "somebot:4@bot".parse().unwrap(), ..Default::default() }) - .message_ids(vec!["MSG001".to_string()]) + .message_ids(vec!["MSG001".into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(false) @@ -4743,7 +4743,7 @@ mod tests { sender: "somebot@bot".parse().unwrap(), ..Default::default() }) - .message_ids(vec!["MSG001".to_string()]) + .message_ids(vec!["MSG001".into()]) .timestamp(wacore::time::now_utc()) .r#type(crate::types::presence::ReceiptType::Retry) .offline(false) diff --git a/tests/e2e/tests/offline_receipts.rs b/tests/e2e/tests/offline_receipts.rs index 9b22c2aa3..10d6be386 100644 --- a/tests/e2e/tests/offline_receipts.rs +++ b/tests/e2e/tests/offline_receipts.rs @@ -31,7 +31,7 @@ async fn test_deferred_delivery_receipt() -> anyhow::Result<()> { matches!( e, Event::Receipt(receipt) - if receipt.message_ids.contains(&msg_id) + if receipt.message_ids.iter().any(|id| *id == msg_id) && receipt.r#type == ReceiptType::Delivered ) }, @@ -83,7 +83,7 @@ async fn test_bidirectional_offline_receipt() -> anyhow::Result<()> { matches!( e, Event::Receipt(receipt) - if receipt.message_ids.contains(&msg_id) + if receipt.message_ids.iter().any(|id| *id == msg_id) && receipt.r#type == ReceiptType::Delivered ) }) @@ -128,7 +128,7 @@ async fn test_deferred_delivery_receipt_on_reconnect() -> anyhow::Result<()> { matches!( e, Event::Receipt(receipt) - if receipt.message_ids.contains(&msg_id) + if receipt.message_ids.iter().any(|id| *id == msg_id) && receipt.r#type == ReceiptType::Delivered ) }, @@ -148,7 +148,7 @@ async fn test_deferred_delivery_receipt_on_reconnect() -> anyhow::Result<()> { matches!( e, Event::Receipt(receipt) - if receipt.message_ids.contains(&msg_id) + if receipt.message_ids.iter().any(|id| *id == msg_id) && receipt.r#type == ReceiptType::Delivered ) }) diff --git a/tests/e2e/tests/receipts.rs b/tests/e2e/tests/receipts.rs index 4bbb1bd57..738444c0e 100644 --- a/tests/e2e/tests/receipts.rs +++ b/tests/e2e/tests/receipts.rs @@ -40,7 +40,7 @@ async fn test_delivery_receipt_online() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Delivered ) }) @@ -83,7 +83,7 @@ async fn test_read_receipt_online() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Read ) }) @@ -123,7 +123,7 @@ async fn test_delivery_receipt_offline_reconnect() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Delivered ) }, @@ -141,7 +141,7 @@ async fn test_delivery_receipt_offline_reconnect() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Delivered ) }) @@ -179,7 +179,7 @@ async fn test_read_receipt_queued_for_offline_sender() -> anyhow::Result<()> { .wait_for_event(10, |e| { matches!( e, - Event::Receipt(r) if r.message_ids.contains(&msg_id) + Event::Receipt(r) if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Delivered ) }) @@ -201,7 +201,7 @@ async fn test_read_receipt_queued_for_offline_sender() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Read ) }) @@ -250,7 +250,7 @@ async fn test_delivery_receipt_bidirectional_offline() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Delivered ) }) @@ -292,7 +292,7 @@ async fn test_no_delivery_receipt_for_fully_offline() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Delivered ) }, @@ -347,7 +347,7 @@ async fn test_group_delivery_receipt() -> anyhow::Result<()> { matches!( e, Event::Receipt(r) - if r.message_ids.contains(&msg_id) + if r.message_ids.iter().any(|id| *id == msg_id) && r.r#type == ReceiptType::Delivered ) }) diff --git a/wacore/binary/src/jid.rs b/wacore/binary/src/jid.rs index be08f2af2..6cafa7857 100644 --- a/wacore/binary/src/jid.rs +++ b/wacore/binary/src/jid.rs @@ -371,7 +371,10 @@ pub const BOT_SERVER: &str = "bot"; pub const STATUS_BROADCAST_USER: &str = "status"; pub const PSA_USER: &str = "0"; -pub type MessageId = String; +/// A message id, inline: WhatsApp ids are 20–32 hex/base64 characters, and +/// every client-generated one is 22, so the common case pays no heap +/// allocation per message, per receipt id and per dedup-cache key. +pub type MessageId = CompactString; pub type MessageServerId = i32; #[derive(Debug)] pub enum JidError { diff --git a/wacore/binary/src/node.rs b/wacore/binary/src/node.rs index 2e8c4d4d5..ecff21682 100644 --- a/wacore/binary/src/node.rs +++ b/wacore/binary/src/node.rs @@ -229,6 +229,13 @@ impl From for NodeValue { } } +impl From<&CompactString> for NodeValue { + #[inline] + fn from(s: &CompactString) -> Self { + NodeValue::String(s.clone()) + } +} + impl From for NodeValue { #[inline] fn from(jid: Jid) -> Self { diff --git a/wacore/src/history_sync.rs b/wacore/src/history_sync.rs index 10d469e57..f7ef015bd 100644 --- a/wacore/src/history_sync.rs +++ b/wacore/src/history_sync.rs @@ -2438,12 +2438,12 @@ mod tests { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![ wa::Conversation { - id: "111222333444555@lid".to_string(), + id: "111222333444555@lid".into(), pn_jid: Some("12025550143@s.whatsapp.net".to_string()), ..Default::default() }, wa::Conversation { - id: "12025550144@s.whatsapp.net".to_string(), + id: "12025550144@s.whatsapp.net".into(), lid_jid: Some("222333444555666@lid".to_string()), ..Default::default() }, @@ -2546,7 +2546,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: "111222333444555@hosted.lid".to_string(), + id: "111222333444555@hosted.lid".into(), pn_jid: Some("12025550143@hosted".to_string()), ..Default::default() }], @@ -2571,7 +2571,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: "111222333444555@lid".to_string(), + id: "111222333444555@lid".into(), pn_jid: Some("12025550143@s.whatsapp.net".to_string()), ..Default::default() }], @@ -2600,7 +2600,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: "111222333444555@lid".to_string(), + id: "111222333444555@lid".into(), pn_jid: Some("12025550143@s.whatsapp.net".to_string()), ..Default::default() }], @@ -2624,7 +2624,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: "15550001111@c.us".to_string(), + id: "15550001111@c.us".into(), lid_jid: Some("222333444555666@lid".to_string()), ..Default::default() }], @@ -2650,7 +2650,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: "999888777666555@lid".to_string(), + id: "999888777666555@lid".into(), pn_jid: Some("12025550143@s.whatsapp.net".to_string()), ..Default::default() }], @@ -2679,7 +2679,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: "222333444555666@lid".to_string(), + id: "222333444555666@lid".into(), pn_jid: Some("12025550144@s.whatsapp.net".to_string()), ..Default::default() }], @@ -3488,7 +3488,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![wa::HistorySyncMsg { message: buffa::MessageField::some(wa::WebMessageInfo { key: buffa::MessageField::some(wa::MessageKey { @@ -3692,7 +3692,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![ wa::HistorySyncMsg { message: buffa::MessageField::some(wa::WebMessageInfo { @@ -3766,7 +3766,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![ wa::HistorySyncMsg { message: buffa::MessageField::some(dropped), @@ -3814,7 +3814,7 @@ mod tests { message.message_secret = Some(vec![0x44; 32]); let hs = wa::HistorySync { conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![wa::HistorySyncMsg { message: buffa::MessageField::some(message), ..Default::default() @@ -3874,7 +3874,7 @@ mod tests { .collect(); let hs = wa::HistorySync { conversations: vec![wa::Conversation { - id: "5511777776666@s.whatsapp.net".to_string(), + id: "5511777776666@s.whatsapp.net".into(), messages, ..Default::default() }], @@ -4018,7 +4018,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![wa::HistorySyncMsg { message: buffa::MessageField::some(wa::WebMessageInfo { key: buffa::MessageField::some(wa::MessageKey { @@ -4067,7 +4067,7 @@ mod tests { let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::INITIAL_BOOTSTRAP, conversations: vec![wa::Conversation { - id: chat.to_string(), + id: chat.into(), messages: vec![wa::HistorySyncMsg { message: buffa::MessageField::some(wa::WebMessageInfo { key: buffa::MessageField::some(wa::MessageKey { @@ -4240,7 +4240,7 @@ mod tests { }); } let big_conv = wa::Conversation { - id: dm.to_string(), + id: dm.into(), messages: big_msgs, tc_token: Some(vec![0xABu8; 16]), tc_token_timestamp: Some(1_700_000_123), @@ -4249,7 +4249,7 @@ mod tests { // Group conversation: a secret message, but its tctoken must be ignored. let group_conv = wa::Conversation { - id: group.to_string(), + id: group.into(), messages: vec![wa::HistorySyncMsg { message: buffa::MessageField::some(wa::WebMessageInfo { key: buffa::MessageField::some(wa::MessageKey { diff --git a/wacore/src/iq/business.rs b/wacore/src/iq/business.rs index 7b543465d..0240933ca 100644 --- a/wacore/src/iq/business.rs +++ b/wacore/src/iq/business.rs @@ -636,7 +636,7 @@ pub struct RemoveCoverPhotoSpec { impl RemoveCoverPhotoSpec { pub fn new(id: &str) -> Self { - Self { id: id.to_string() } + Self { id: id.into() } } } diff --git a/wacore/src/messages.rs b/wacore/src/messages.rs index eea9d562f..a96ed981e 100644 --- a/wacore/src/messages.rs +++ b/wacore/src/messages.rs @@ -1190,7 +1190,7 @@ pub fn parse_message_info( !id.is_empty(), "message stanza has an empty required 'id' attribute" ); - let id = id.into_owned(); + let id = CompactString::from(id.as_ref()); let from = attrs.required_jid("from")?; let addressing_mode = attrs .optional_string("addressing_mode") @@ -1340,8 +1340,11 @@ pub fn parse_message_info( // child attrs (WAWebHandleMsgParser b()) and children // (I() function). Both are optional; absence is the common case. - let mut meta_info = crate::types::message::MsgMetaInfo::default(); + // Built only when one of the children is present, so the common message + // carries no `MsgMetaInfo` at all. + let mut meta_info: Option> = None; if let Some(meta) = node.get_optional_child("meta") { + let meta_info = meta_info.get_or_insert_with(Default::default); let mut ma = meta.attrs(); meta_info.content_type = ma.optional_string("content_type").map(CompactString::from); meta_info.appdata = ma.optional_string("appdata").map(CompactString::from); @@ -1364,11 +1367,13 @@ pub fn parse_message_info( if let Some(reporting) = node.get_optional_child("reporting") && let Some(tag) = reporting.get_optional_child("reporting_tag") { + let meta_info = meta_info.get_or_insert_with(Default::default); meta_info.reporting_tag = tag.content_bytes().map(ReportingBytes::from_slice); } if let Some(reporting) = node.get_optional_child("reporting") && let Some(token) = reporting.get_optional_child("reporting_token") { + let meta_info = meta_info.get_or_insert_with(Default::default); meta_info.reporting_token = token.content_bytes().map(ReportingBytes::from_slice); // WA Web `I()`: `c.maybeAttrInt("v")!=null?_:1`. Missing `v` is // not a parse failure — token format version defaults to 1. @@ -1387,16 +1392,18 @@ pub fn parse_message_info( // but parsing it always is a strict superset. let bot_info = node.get_optional_child("bot").map(|bot_node| { let mut ba = bot_node.attrs(); - crate::types::message::MsgBotInfo { + Box::new(crate::types::message::MsgBotInfo { edit_type: ba .optional_string("edit") .and_then(|s| crate::types::message::BotEditType::from_wire(s.as_ref())), - edit_target_id: ba.optional_string("edit_target_id").map(|s| s.into_owned()), + edit_target_id: ba + .optional_string("edit_target_id") + .map(|s| CompactString::from(s.as_ref())), edit_sender_timestamp_ms: ba .optional_u64("sender_timestamp_ms") .and_then(|ms| i64::try_from(ms).ok()) .and_then(crate::time::from_millis), - } + }) }); Ok(MessageInfo { @@ -1406,7 +1413,7 @@ pub fn parse_message_info( r#type: stanza_type, push_name: attrs .optional_string("notify") - .map(|s| s.to_string()) + .map(|s| CompactString::from(s.as_ref())) .unwrap_or_default(), timestamp: crate::time::from_secs_or_now(attrs.unix_time("t")), category, @@ -1804,8 +1811,8 @@ mod parse_message_info_tests { .build()]) .build(); let info = parse_message_info(&node.as_node_ref(), &own_pn, None).unwrap(); - assert_eq!(info.meta_info.content_type.as_deref(), Some("add_on")); - assert!(info.meta_info.appdata.is_none()); + assert_eq!(info.meta().content_type.as_deref(), Some("add_on")); + assert!(info.meta().appdata.is_none()); } /// `{bytes} @@ -1835,14 +1842,14 @@ mod parse_message_info_tests { .build(); let info = parse_message_info(&node.as_node_ref(), &own_pn, None).unwrap(); assert_eq!( - info.meta_info.reporting_tag.as_deref(), + info.meta().reporting_tag.as_deref(), Some(tag_bytes.as_slice()) ); assert_eq!( - info.meta_info.reporting_token.as_deref(), + info.meta().reporting_token.as_deref(), Some(token_bytes.as_slice()) ); - assert_eq!(info.meta_info.reporting_token_version, Some(2)); + assert_eq!(info.meta().reporting_token_version, Some(2)); } /// Missing `v` attr on `` defaults the version to 1 @@ -1862,7 +1869,7 @@ mod parse_message_info_tests { .build()]) .build(); let info = parse_message_info(&node.as_node_ref(), &own_pn, None).unwrap(); - assert_eq!(info.meta_info.reporting_token_version, Some(1)); + assert_eq!(info.meta().reporting_token_version, Some(1)); } /// `` with ONLY `` (no token) is also valid @@ -1882,9 +1889,9 @@ mod parse_message_info_tests { .build()]) .build(); let info = parse_message_info(&node.as_node_ref(), &own_pn, None).unwrap(); - assert!(info.meta_info.reporting_tag.is_some()); - assert!(info.meta_info.reporting_token.is_none()); - assert!(info.meta_info.reporting_token_version.is_none()); + assert!(info.meta().reporting_tag.is_some()); + assert!(info.meta().reporting_token.is_none()); + assert!(info.meta().reporting_token_version.is_none()); } /// Message with no `` and no `` leaves all the new @@ -1899,10 +1906,10 @@ mod parse_message_info_tests { .attr("t", "1777415965") .build(); let info = parse_message_info(&node.as_node_ref(), &own_pn, None).unwrap(); - assert!(info.meta_info.content_type.is_none()); - assert!(info.meta_info.appdata.is_none()); - assert!(info.meta_info.reporting_tag.is_none()); - assert!(info.meta_info.reporting_token.is_none()); + assert!(info.meta().content_type.is_none()); + assert!(info.meta().appdata.is_none()); + assert!(info.meta().reporting_tag.is_none()); + assert!(info.meta().reporting_token.is_none()); } /// Symmetric branch: when `participant` is a LID, `sender_alt` must come @@ -2193,12 +2200,13 @@ mod parse_message_info_tests { let poll = with_meta("poll"); assert_eq!(poll.r#type, Some(T::Poll)); - assert_eq!(poll.meta_info.poll_type, Some(PollType::Vote)); + assert_eq!(poll.meta().poll_type, Some(PollType::Vote)); let text = with_meta("text"); assert_eq!(text.r#type, Some(T::Text)); assert_eq!( - text.meta_info.poll_type, None, + text.meta().poll_type, + None, "polltype belongs to poll envelopes only" ); } @@ -2216,7 +2224,7 @@ mod parse_message_info_tests { .attr("polltype", "retraction") .build()]) .build(); - assert_eq!(parse(&node).meta_info.poll_type, None); + assert_eq!(parse(&node).meta().poll_type, None); } #[test] @@ -2233,12 +2241,9 @@ mod parse_message_info_tests { .build()]) .build(); let info = parse(&node); + assert_eq!(info.meta().thread_message_id.as_deref(), Some("PARENT-1")); assert_eq!( - info.meta_info.thread_message_id.as_deref(), - Some("PARENT-1") - ); - assert_eq!( - info.meta_info + info.meta() .thread_message_sender_jid .as_ref() .map(|jid| jid.user.as_str()), diff --git a/wacore/src/stanza/business.rs b/wacore/src/stanza/business.rs index cb085f873..39d62e74c 100644 --- a/wacore/src/stanza/business.rs +++ b/wacore/src/stanza/business.rs @@ -322,7 +322,7 @@ impl BusinessNotification { child.attrs().optional_string("status"), ) { subscriptions.push(BusinessSubscription { - id: id.to_string(), + id: id.into(), status: status.to_string(), expiration_date: child .attrs() diff --git a/wacore/src/stanza/receipt.rs b/wacore/src/stanza/receipt.rs index 94ada65da..4fd428ba5 100644 --- a/wacore/src/stanza/receipt.rs +++ b/wacore/src/stanza/receipt.rs @@ -85,18 +85,22 @@ pub fn parse_participants( /// cloned into the vector and dropped one line later. pub fn collect_simple_message_ids( node: &NodeRef<'_>, - stanza_id: String, + stanza_id: wacore_binary::MessageId, is_view: bool, -) -> Vec { +) -> Vec { let id_attr = if is_view { "server_id" } else { "id" }; - let mut ids: Vec = node + let mut ids: Vec = node .get_optional_child("list") .and_then(|list| { list.children().map(|items| { items .iter() .filter(|c| c.tag == "item") - .filter_map(|c| c.attrs().optional_string(id_attr).map(|s| s.into_owned())) + .filter_map(|c| { + c.attrs() + .optional_string(id_attr) + .map(|s| wacore_binary::MessageId::from(s.as_ref())) + }) .collect() }) }) @@ -222,7 +226,7 @@ mod tests { #[test] fn skip_empty_id() { let info = MessageInfo { - id: "".to_string(), + id: "".into(), source: MessageSource { chat: "12345@s.whatsapp.net".parse().unwrap(), sender: "12345@s.whatsapp.net".parse().unwrap(), @@ -237,7 +241,7 @@ mod tests { #[test] fn skip_status_broadcast() { let info = MessageInfo { - id: "MSG1".to_string(), + id: "MSG1".into(), source: MessageSource { chat: "status@broadcast".parse().unwrap(), sender: "12345@s.whatsapp.net".parse().unwrap(), @@ -253,7 +257,7 @@ mod tests { #[test] fn skip_newsletter() { let info = MessageInfo { - id: "NL1".to_string(), + id: "NL1".into(), source: MessageSource { chat: "120363173003902460@newsletter".parse().unwrap(), sender: "120363173003902460@newsletter".parse().unwrap(), @@ -268,7 +272,7 @@ mod tests { #[test] fn skip_own_non_peer_messages() { let info = MessageInfo { - id: "OWN1".to_string(), + id: "OWN1".into(), source: MessageSource { chat: "12345@s.whatsapp.net".parse().unwrap(), sender: "12345@s.whatsapp.net".parse().unwrap(), @@ -283,7 +287,7 @@ mod tests { #[test] fn allow_peer_self_synced_messages() { let info = MessageInfo { - id: "PEER1".to_string(), + id: "PEER1".into(), source: MessageSource { chat: "12345@s.whatsapp.net".parse().unwrap(), sender: "12345@s.whatsapp.net".parse().unwrap(), @@ -302,7 +306,7 @@ mod tests { // sender receipt. A recipient-less own message (skip_own_non_peer_*) // stays skipped. Mirrors the hot-path copy in the whatsapp-rust crate. let info = MessageInfo { - id: "FANOUT1".to_string(), + id: "FANOUT1".into(), source: MessageSource { chat: "200000000000002@bot".parse().unwrap(), sender: "100000000000001@lid".parse().unwrap(), @@ -321,7 +325,7 @@ mod tests { // self-fanout allowance must NOT leak into own status broadcasts or // group messages, even when a recipient is present. let own_status = MessageInfo { - id: "OWN_STATUS".to_string(), + id: "OWN_STATUS".into(), source: MessageSource { chat: "status@broadcast".parse().unwrap(), sender: "100000000000001@lid".parse().unwrap(), @@ -334,7 +338,7 @@ mod tests { assert!(!should_send_delivery_receipt(&own_status)); let own_group = MessageInfo { - id: "OWN_GROUP".to_string(), + id: "OWN_GROUP".into(), source: MessageSource { chat: "120363021033254949@g.us".parse().unwrap(), sender: "100000000000001@lid".parse().unwrap(), @@ -351,7 +355,7 @@ mod tests { #[test] fn allow_incoming_dm() { let info = MessageInfo { - id: "DM1".to_string(), + id: "DM1".into(), source: MessageSource { chat: "12345@s.whatsapp.net".parse().unwrap(), sender: "12345@s.whatsapp.net".parse().unwrap(), @@ -464,7 +468,7 @@ mod tests { ]) .build()]) .build(); - let ids = collect_simple_message_ids(&node.as_node_ref(), "STANZA-Z".to_string(), false); + let ids = collect_simple_message_ids(&node.as_node_ref(), "STANZA-Z".into(), false); assert_eq!(ids, vec!["MSG-A", "MSG-B", "STANZA-Z"]); } @@ -472,7 +476,7 @@ mod tests { #[test] fn simple_message_ids_without_list() { let node = NodeBuilder::new("receipt").attr("id", "SOLO").build(); - let ids = collect_simple_message_ids(&node.as_node_ref(), "SOLO".to_string(), false); + let ids = collect_simple_message_ids(&node.as_node_ref(), "SOLO".into(), false); assert_eq!(ids, vec!["SOLO"]); } @@ -489,7 +493,7 @@ mod tests { ]) .build()]) .build(); - let ids = collect_simple_message_ids(&node.as_node_ref(), "VIEW-STANZA".to_string(), true); + let ids = collect_simple_message_ids(&node.as_node_ref(), "VIEW-STANZA".into(), true); assert_eq!(ids, vec!["100", "101"]); } @@ -505,7 +509,7 @@ mod tests { #[test] fn listless_receipt_ids_are_sized_to_what_they_hold() { let node = NodeBuilder::new("receipt").attr("id", "SOLO").build(); - let ids = collect_simple_message_ids(&node.as_node_ref(), "SOLO".to_string(), false); + let ids = collect_simple_message_ids(&node.as_node_ref(), "SOLO".into(), false); assert_eq!(ids.len(), 1); assert!( ids.capacity() < 4, @@ -524,7 +528,7 @@ mod tests { .children([NodeBuilder::new("item").attr("id", "MSG-A").build()]) .build()]) .build(); - let ids = collect_simple_message_ids(&node.as_node_ref(), "VIEW-STANZA".to_string(), true); + let ids = collect_simple_message_ids(&node.as_node_ref(), "VIEW-STANZA".into(), true); assert!( ids.is_empty(), "items without server_id contribute nothing and the stanza id is not appended" diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 9cea95428..50b45622f 100755 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -1245,6 +1245,18 @@ impl fmt::Debug for Event { pub struct InboundMessage { pub message: Arc, pub info: Arc, + /// Ephemeral duration in seconds, from the decrypted message's + /// `contextInfo.expiration`. Lives here rather than on `info` because it + /// is only known after decryption, and `info` is shared with every + /// `` of the stanza by then: writing it there cost a deep copy of + /// the whole `MessageInfo` on every disappearing-chat message. + #[serde(skip_serializing_if = "Option::is_none")] + pub ephemeral_expiration: Option, + /// Parent post key when `message` is a decrypted CAG channel comment + /// (`enc_comment_message`). The inner `Message` proto has no slot for the + /// threading link, so it surfaces here. Boxed: rare. + #[serde(skip_serializing_if = "Option::is_none")] + pub comment_target: Option>, } /// How a [`MessageBatch`] was delivered. This describes the delivery shape, @@ -2722,7 +2734,7 @@ mod tests { #[test] fn lazy_history_sync_get_decodes() { let lazy = lazy_from(vec![wa::Conversation { - id: "chat@s.whatsapp.net".to_string(), + id: "chat@s.whatsapp.net".into(), ..Default::default() }]); @@ -2734,7 +2746,7 @@ mod tests { #[test] fn lazy_history_sync_caches_decode() { let lazy = lazy_from(vec![wa::Conversation { - id: "test@g.us".to_string(), + id: "test@g.us".into(), ..Default::default() }]); @@ -2775,7 +2787,7 @@ mod tests { #[test] fn lazy_history_sync_decompress_yields_raw_proto() { let lazy = lazy_from(vec![wa::Conversation { - id: "raw@s.whatsapp.net".to_string(), + id: "raw@s.whatsapp.net".into(), ..Default::default() }]); @@ -2792,7 +2804,7 @@ mod tests { #[test] fn lazy_history_sync_everything_keeps_working_after_get() { let lazy = lazy_from(vec![wa::Conversation { - id: "kept@s.whatsapp.net".to_string(), + id: "kept@s.whatsapp.net".into(), ..Default::default() }]); @@ -2817,11 +2829,11 @@ mod tests { fn lazy_history_sync_stream_iterates_conversations() { let lazy = lazy_from(vec![ wa::Conversation { - id: "first@s.whatsapp.net".to_string(), + id: "first@s.whatsapp.net".into(), ..Default::default() }, wa::Conversation { - id: "second@s.whatsapp.net".to_string(), + id: "second@s.whatsapp.net".into(), ..Default::default() }, ]); @@ -2847,7 +2859,7 @@ mod tests { #[test] fn lazy_history_sync_clone_is_cheap_and_redecodes() { let lazy = lazy_from(vec![wa::Conversation { - id: "cloned@s.whatsapp.net".to_string(), + id: "cloned@s.whatsapp.net".into(), ..Default::default() }]); @@ -2892,7 +2904,7 @@ mod tests { // A decompressed_size below the real inflated size trips the inflate // cap instead of silently over-allocating past the producer's count. let (compressed, raw_len) = make_compressed_history_sync(vec![wa::Conversation { - id: "capped@s.whatsapp.net".to_string(), + id: "capped@s.whatsapp.net".into(), ..Default::default() }]); let lazy = LazyHistorySync::new(compressed, raw_len - 1, 0, None, None); @@ -2903,7 +2915,7 @@ mod tests { #[test] fn lazy_history_sync_preserves_messages() { let conv = wa::Conversation { - id: "chat@s.whatsapp.net".to_string(), + id: "chat@s.whatsapp.net".into(), messages: vec![wa::HistorySyncMsg { message: wa::WebMessageInfo { key: wa::MessageKey { diff --git a/wacore/src/types/message.rs b/wacore/src/types/message.rs index 30a972526..3fd40fc63 100644 --- a/wacore/src/types/message.rs +++ b/wacore/src/types/message.rs @@ -1,7 +1,6 @@ use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; use wacore_binary::{CompactString, Jid, JidExt, MessageId, MessageServerId}; -use waproto::whatsapp as wa; use crate::WireEnum; use smallvec::SmallVec; @@ -398,7 +397,9 @@ pub struct MessageInfo { /// The envelope's `type` attribute. `None` when the stanza carried none. #[serde(skip_serializing_if = "Option::is_none")] pub r#type: Option, - pub push_name: String, + /// The sender's `notify` display name. Inline up to 24 bytes, which + /// covers most names, so a message does not allocate for it. + pub push_name: CompactString, #[serde(serialize_with = "chrono::serde::ts_seconds::serialize")] pub timestamp: DateTime, pub category: MessageCategory, @@ -424,18 +425,22 @@ pub struct MessageInfo { #[serde(skip_serializing_if = "Option::is_none")] pub media_type: Option, pub edit: EditAttribute, + /// The `` child. Boxed: most messages carry none. #[serde(skip_serializing_if = "Option::is_none")] - pub bot_info: Option, - pub meta_info: MsgMetaInfo, + pub bot_info: Option>, + /// The `` and `` children, `None` when the stanza carries + /// neither. Boxed: it is 280 bytes of mostly-absent fields, and every + /// `MessageInfo` is retained per message through the commit batch and + /// every consumer that keeps a message. + #[serde(skip_serializing_if = "Option::is_none")] + pub meta_info: Option>, /// Decoded `` child cert of business senders; the display /// name is in `.name`. Boxed: most messages carry none. #[serde(skip_serializing_if = "Option::is_none")] pub verified_name: Option>, + /// Set on a self-fanout of an own outgoing message. Boxed: rare. #[serde(skip_serializing_if = "Option::is_none")] - pub device_sent_meta: Option, - /// Ephemeral duration in seconds, extracted from `contextInfo.expiration`. - #[serde(skip_serializing_if = "Option::is_none")] - pub ephemeral_expiration: Option, + pub device_sent_meta: Option>, /// Whether this message was delivered during offline sync. pub is_offline: bool, /// Set when this message was recovered via PDO rather than normal decryption. @@ -461,11 +466,6 @@ pub struct MessageInfo { /// goes to the right routing target). #[serde(skip_serializing_if = "Option::is_none")] pub peer_recipient_pn: Option, - /// Parent post key when the dispatched message is a decrypted CAG channel - /// comment (`enc_comment_message`). The inner `Message` proto has no slot - /// for the threading link, so it surfaces here. - #[serde(skip_serializing_if = "Option::is_none")] - pub comment_target: Option, /// Broadcast-contact-list recipients from `` on an /// incoming broadcast/status stanza. Populated only for broadcasts; used to /// validate a `deviceSentMessage.phash` (WA Web `validateBclHash`). Empty @@ -474,7 +474,20 @@ pub struct MessageInfo { pub bcl_participants: Vec, } +/// What [`MessageInfo::meta`] hands out for a stanza that carried no `` +/// or `` child: every field `None`, shared by every such message. +static EMPTY_META: std::sync::LazyLock = + std::sync::LazyLock::new(MsgMetaInfo::default); + impl MessageInfo { + /// The `` and `` data, or an all-`None` one when the + /// stanza carried neither. Readers that only look at a field go through + /// here; [`meta_info`](Self::meta_info) itself is `None` in that case so + /// the common message does not allocate it. + pub fn meta(&self) -> &MsgMetaInfo { + self.meta_info.as_deref().unwrap_or(&EMPTY_META) + } + /// WA Web: expired status messages (>24h) are silently dropped — no retry receipts, /// no undecryptable events. Matches `WAWebMsgProcessingDecryptionHandler.E()`. pub fn is_expired_status(&self) -> bool { @@ -514,7 +527,10 @@ mod tests { fn message_info_serde_omits_only_absent_optional_fields() { let mut info = MessageInfo::default(); info.source.sender_alt = Some("15550000001@lid".parse().unwrap()); - info.meta_info.target_id = Some("TARGET".into()); + info.meta_info = Some(Box::new(MsgMetaInfo { + target_id: Some("TARGET".into()), + ..Default::default() + })); info.unavailable_request_id = Some("REQUEST".to_owned()); let serialized = serde_json::to_value(info).expect("serialize message info"); @@ -538,7 +554,6 @@ mod tests { assert!(!root.contains_key("bot_info")); assert!(!root.contains_key("verified_name")); assert!(!root.contains_key("device_sent_meta")); - assert!(!root.contains_key("ephemeral_expiration")); assert_eq!( root.get("timestamp").and_then(|value| value.as_i64()), Some(0) From 01c6da55331d9084c05716d76ed8d2c1fcbcc961 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 17:43:06 +0000 Subject: [PATCH 3/4] chore(signal): keep checkout results unboxed now that SessionRecord grew `SessionState` carries its skipped-key backlog beside the protobuf, which takes `SessionRecord` past clippy's `large_enum_variant` threshold in the two checkout-result enums that carry one. Both are transient return values matched and moved out of by their caller on the per-message path, so boxing the record arm would add an allocation per checkout to save nothing; the lint is allowed at each with that rationale. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_0172fpxasGTrouFyYH5UGmjN --- wacore/libsignal/src/protocol/storage/traits.rs | 6 ++++++ wacore/src/store/signal_cache.rs | 4 ++++ 2 files changed, 10 insertions(+) diff --git a/wacore/libsignal/src/protocol/storage/traits.rs b/wacore/libsignal/src/protocol/storage/traits.rs index ddf84cd38..f107d16a9 100644 --- a/wacore/libsignal/src/protocol/storage/traits.rs +++ b/wacore/libsignal/src/protocol/storage/traits.rs @@ -291,7 +291,13 @@ pub trait SessionStore: ThreadSafe { } /// Result of returning a record from a cancellation-safe checkout. +/// +/// A transient return value, moved out of as soon as it is matched, so the +/// record-carrying arm costs nothing beyond the record itself; boxing it +/// would add an allocation to a path that is done with the value by the next +/// statement. #[doc(hidden)] +#[allow(clippy::large_enum_variant)] pub enum SessionCheckoutStoreResult { Stored, Rejected, diff --git a/wacore/src/store/signal_cache.rs b/wacore/src/store/signal_cache.rs index 295b866f9..68c9961e9 100644 --- a/wacore/src/store/signal_cache.rs +++ b/wacore/src/store/signal_cache.rs @@ -472,6 +472,10 @@ impl SessionEntry { } } +/// A transient return value, matched and moved out of by its caller on the +/// per-message decrypt path; boxing the record arm would put an allocation on +/// every checkout to save nothing, since the enum is never stored. +#[allow(clippy::large_enum_variant)] enum CachedSessionCheckout { Missing(SessionCheckoutKey), Absent(SessionCheckoutKey), From 5e06bf50b46ab7933c6eca34fdd15a88b4180e93 Mon Sep 17 00:00:00 2001 From: Claude Date: Wed, 2 Sep 2026 17:43:06 +0000 Subject: [PATCH 4/4] perf(client): classify a stanza once per frame and release burst tables `process_node` derived the `StanzaTag` for its dispatch match but each gate before it (offline-sync banner, IQ sync response, stream end, IQ waiter) re-compared the tag string, and the match itself was a chain of string guards. The tag is now derived once and every gate dispatches on the enum. `pending_retries` and `pending_lid_refreshes` hold one entry per in-flight operation and are empty almost all the time, but a reconnect can push hundreds of retries through at once and a `HashSet` never gives that table back. The scopeguard that removes a reservation now shrinks the set once it is a quarter full, to twice its length so a draining burst does not oscillate between shrink and regrow. Co-Authored-By: Claude Fable 5.1 Claude-Session: https://claude.ai/code/session_0172fpxasGTrouFyYH5UGmjN --- src/client.rs | 14 ++++++++++++++ src/client/lid_pn.rs | 7 +++---- src/client/node_io.rs | 29 ++++++++++++++--------------- src/retry.rs | 7 +++---- 4 files changed, 34 insertions(+), 23 deletions(-) diff --git a/src/client.rs b/src/client.rs index 546725a25..482cc613f 100644 --- a/src/client.rs +++ b/src/client.rs @@ -2181,5 +2181,19 @@ fn fibonacci_backoff(attempt: u32) -> Duration { Duration::from_millis(ms) } +/// Release the table a reservation set grew during a burst. +/// +/// `pending_retries` and `pending_lid_refreshes` hold one entry per in-flight +/// operation and are empty almost all the time, but a reconnect can push +/// hundreds of retries through at once and a `HashSet` never gives that table +/// back on its own. The `len * 4` threshold keeps a set that is still draining +/// from oscillating between shrink and regrow; `shrink_to` rather than +/// `shrink_to_fit` leaves room for the tail of the burst. +pub(crate) fn release_after_burst(set: &mut HashSet) { + if set.capacity() > 32 && set.len() * 4 < set.capacity() { + set.shrink_to(set.len() * 2); + } +} + #[cfg(test)] mod tests; diff --git a/src/client/lid_pn.rs b/src/client/lid_pn.rs index ea1e6f887..f7236e215 100644 --- a/src/client/lid_pn.rs +++ b/src/client/lid_pn.rs @@ -1335,10 +1335,9 @@ impl Client { } let pending = Arc::clone(&self.pending_lid_refreshes); let _guard = scopeguard::guard((), move |()| { - pending - .lock() - .unwrap_or_else(|p| p.into_inner()) - .remove(&key); + let mut pending = pending.lock().unwrap_or_else(|p| p.into_inner()); + pending.remove(&key); + super::release_after_burst(&mut pending); }); // Persists through `add_lid_pn_mapping`, so a corrected pair is durable diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 708917b7b..614c276f2 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -377,9 +377,12 @@ impl Client { pub(crate) async fn process_node(self: &Arc, node: Arc) { use wacore::xml::DisplayableNodeRef; let nr = node.get(); + // Classified once; every gate below dispatches on the enum instead of + // re-comparing the tag string. + let tag = StanzaTag::try_from(nr.tag.as_ref()).ok(); // --- Offline Sync Tracking --- - if nr.tag.as_ref() == StanzaTag::InfoBanner.as_str() { + if tag == Some(StanzaTag::InfoBanner) { // Check for offline_preview child to get expected count if let Some(preview) = nr.get_optional_child("offline_preview") { let count: usize = preview @@ -469,7 +472,7 @@ impl Client { } // --- End Tracking --- - if nr.tag.as_ref() == StanzaTag::Iq.as_str() + if tag == Some(StanzaTag::Iq) && let Some(sync_node) = nr.get_optional_child("sync") && let Some(collection_node) = sync_node.get_optional_child("collection") { @@ -491,7 +494,7 @@ impl Client { .dispatch(Event::RawNode(Arc::clone(&node))); } - if nr.tag.as_ref() == StanzaTag::XmlStreamEnd.as_str() { + if tag == Some(StanzaTag::XmlStreamEnd) { if self.expected_disconnect.load(Ordering::Relaxed) { debug!("Received , expected disconnect."); } else { @@ -508,7 +511,7 @@ impl Client { self.resolve_node_waiters(&node); } - if nr.tag.as_ref() == StanzaTag::Iq.as_str() + if tag == Some(StanzaTag::Iq) && let Some(id) = nr.get_attr("id").map(|v| v.as_str()) && let Some(waiter) = self.response_waiters_guard().remove(id.as_ref()) { @@ -574,14 +577,14 @@ impl Client { // Bypass async_trait's boxed future for the hot built-in handlers while // retaining router registration for direct router callers. - match nr.tag.as_ref() { - t if t == StanzaTag::Ack.as_str() => { + match tag { + Some(StanzaTag::Ack) => { self.handle_ack_response_arc(&node); } - t if t == StanzaTag::Receipt.as_str() => { + Some(StanzaTag::Receipt) => { self.handle_receipt_inline(node); } - t if t == StanzaTag::Message.as_str() => { + Some(StanzaTag::Message) => { crate::handlers::message::MessageHandler::handle_inline( self.clone(), node, @@ -591,7 +594,7 @@ impl Client { } // Differs from a `` only in tag, so WA Web retags it and // runs the same pipeline. - t if t == StanzaTag::Status.as_str() && is_status_broadcast_stanza(nr) => { + Some(StanzaTag::Status) if is_status_broadcast_stanza(nr) => { crate::handlers::message::MessageHandler::handle_inline( self.clone(), node, @@ -702,14 +705,10 @@ impl Client { /// would redeliver indefinitely. WA Web emits `` /// in the success path on top of this; the duplicate is tolerated. pub(crate) fn should_ack(&self, node: &wacore_binary::NodeRef<'_>) -> bool { - let tag = StanzaTag::try_from(node.tag.as_ref()); - if node.get_attr("id").is_none() { - return false; - } - if node.get_attr("from").is_none() { + if node.get_attr("id").is_none() || node.get_attr("from").is_none() { return false; } - match tag { + match StanzaTag::try_from(node.tag.as_ref()) { Ok(StanzaTag::Receipt | StanzaTag::Notification | StanzaTag::Call) => true, Ok(StanzaTag::Message) => { from_jid_matches(node, |j| j.is_newsletter() || j.is_status_broadcast()) diff --git a/src/retry.rs b/src/retry.rs index 91d29e84a..8a390438d 100644 --- a/src/retry.rs +++ b/src/retry.rs @@ -512,10 +512,9 @@ impl Client { // the scopeguard instead of cloning again. let pending = Arc::clone(&self.pending_retries); let _guard = scopeguard::guard((), move |()| { - pending - .lock() - .unwrap_or_else(|p| p.into_inner()) - .remove(&processing_key); + let mut pending = pending.lock().unwrap_or_else(|p| p.into_inner()); + pending.remove(&processing_key); + crate::client::release_after_burst(&mut pending); }); // A retry from a device missing from our registry signals a stale device