diff --git a/src/client.rs b/src/client.rs index 95fa1f527..4dcdddfdd 100644 --- a/src/client.rs +++ b/src/client.rs @@ -132,6 +132,11 @@ use thiserror::Error; use wacore::appstate::patch_decode::WAPatchName; use wacore::client::context::GroupInfo; + +/// Group metadata cache. Values are `Arc`-wrapped so a warm `query_info` hit +/// shares the metadata (refcount bump) instead of deep-cloning the participant +/// list and LID/PN maps on every group send. +type GroupCache = TypedCache>; use wacore::runtime::timeout as rt_timeout; use waproto::whatsapp as wa; @@ -395,7 +400,7 @@ pub struct Client { pub(crate) lid_pn_cache: Arc, pub(crate) ab_props: Arc, - pub group_cache: async_lock::Mutex>>>, + pub group_cache: async_lock::Mutex>>, pub(crate) expected_disconnect: Arc, /// Set by `reconnect()` to suppress the "Message loop exited with an error" warning. @@ -935,7 +940,7 @@ impl Client { (arc, rx) } - pub(crate) async fn get_group_cache(&self) -> Arc> { + pub(crate) async fn get_group_cache(&self) -> Arc { let mut guard = self.group_cache.lock().await; if let Some(cache) = guard.as_ref() { return cache.clone(); diff --git a/src/client/context_impl.rs b/src/client/context_impl.rs index 32bbb12a3..21ab95dfe 100644 --- a/src/client/context_impl.rs +++ b/src/client/context_impl.rs @@ -1,6 +1,7 @@ use crate::client::Client; use async_trait::async_trait; use std::collections::HashMap; +use std::sync::Arc; use wacore::client::context::{GroupInfo, SendContextResolver}; use wacore::iq::prekeys::PreKeyFetchReason; use wacore::libsignal::protocol::PreKeyBundle; @@ -41,7 +42,7 @@ impl SendContextResolver for Client { }) } - async fn resolve_group_info(&self, jid: &Jid) -> Result { + async fn resolve_group_info(&self, jid: &Jid) -> Result, anyhow::Error> { self.groups().query_info(jid).await } diff --git a/src/features/groups.rs b/src/features/groups.rs index 57339411b..320f76d63 100644 --- a/src/features/groups.rs +++ b/src/features/groups.rs @@ -1,6 +1,7 @@ use crate::client::Client; use crate::features::mex::{MexError, MexRequest}; use std::collections::HashMap; +use std::sync::Arc; use wacore::client::context::GroupInfo; use wacore::iq::groups::{ AcceptGroupInviteIq, AcceptGroupInviteV4Iq, AcknowledgeGroupIq, AddParticipantsIq, @@ -187,7 +188,7 @@ impl<'a> Groups<'a> { Self { client } } - pub async fn query_info(&self, jid: &Jid) -> Result { + pub async fn query_info(&self, jid: &Jid) -> Result, anyhow::Error> { if let Some(cached) = self.client.get_group_cache().await.get(jid).await { return Ok(cached); } @@ -211,9 +212,9 @@ impl<'a> Groups<'a> { .await? { GroupInfoOutcome::NotModified => { - let info = persisted.ok_or_else(|| { + let info = Arc::new(persisted.ok_or_else(|| { anyhow::anyhow!("server returned not-modified group but nothing was cached") - })?; + })?); self.client .get_group_cache() .await @@ -280,6 +281,7 @@ impl<'a> Groups<'a> { Err(e) => log::warn!("Failed to serialize group metadata for {jid}: {e}"), } + let info = Arc::new(info); self.client .get_group_cache() .await @@ -406,14 +408,15 @@ impl<'a> Groups<'a> { let result = self.client.execute(iq).await?; if result.iter().any(|r| r.is_ok()) { let group_cache = self.client.get_group_cache().await; - if let Some(mut info) = group_cache.get(jid).await { + if let Some(info) = group_cache.get(jid).await { + let mut info = Arc::unwrap_or_clone(info); info.add_participants( result .iter() .filter(|r| r.is_ok()) .map(|r| (&r.jid, r.phone_number.as_ref())), ); - group_cache.insert(jid.clone(), info).await; + group_cache.insert(jid.clone(), Arc::new(info)).await; } } Ok(result) @@ -435,9 +438,10 @@ impl<'a> Groups<'a> { .collect(); if !accepted.is_empty() { let group_cache = self.client.get_group_cache().await; - if let Some(mut info) = group_cache.get(jid).await { + if let Some(info) = group_cache.get(jid).await { + let mut info = Arc::unwrap_or_clone(info); info.remove_participants(&accepted); - group_cache.insert(jid.clone(), info).await; + group_cache.insert(jid.clone(), Arc::new(info)).await; } self.client .rotate_sender_key_on_participant_remove(&jid.to_string(), &accepted) @@ -1081,5 +1085,32 @@ mod tests { assert!(extract_invite_code("whatsapp://chat/?code=&other=1").is_none()); } + #[tokio::test] + async fn warm_group_cache_hit_shares_arc_not_deep_clone() { + use wacore::client::context::GroupInfo; + use wacore::types::message::AddressingMode; + + let client = crate::test_utils::create_test_client().await; + let group_jid: Jid = "123456789@g.us".parse().unwrap(); + + let info = GroupInfo::new( + vec![ + "111111111111@s.whatsapp.net".parse().unwrap(), + "222222222222@s.whatsapp.net".parse().unwrap(), + ], + AddressingMode::Pn, + ); + let cache = client.get_group_cache().await; + cache.insert(group_jid.clone(), Arc::new(info)).await; + + let a = cache.get(&group_jid).await.expect("warm hit"); + let b = cache.get(&group_jid).await.expect("warm hit"); + + // A warm group-cache hit returns a refcount bump of the same allocation, + // not a deep copy of the participant list and LID/PN maps. + assert!(Arc::ptr_eq(&a, &b)); + assert_eq!(a.participants.len(), 2); + } + // Protocol-level tests (node building, parsing, validation) are in wacore/src/iq/groups.rs } diff --git a/src/handlers/notification.rs b/src/handlers/notification.rs index 846395a78..8952454d8 100755 --- a/src/handlers/notification.rs +++ b/src/handlers/notification.rs @@ -1181,14 +1181,15 @@ async fn handle_group_notification(client: &Arc, node: Arc match &action { GroupNotificationAction::Add { participants, .. } => { let group_cache = client.get_group_cache().await; - if let Some(mut info) = group_cache.get(¬ification.group_jid).await { + if let Some(info) = group_cache.get(¬ification.group_jid).await { + let mut info = Arc::unwrap_or_clone(info); info.add_participants( participants .iter() .map(|p| (&p.jid, p.phone_number.as_ref())), ); group_cache - .insert(notification.group_jid.clone(), info) + .insert(notification.group_jid.clone(), Arc::new(info)) .await; debug!( target: "Client/Group", @@ -1200,10 +1201,11 @@ async fn handle_group_notification(client: &Arc, node: Arc GroupNotificationAction::Remove { participants, .. } => { let users: Vec<&str> = participants.iter().map(|p| p.jid.user.as_str()).collect(); let group_cache = client.get_group_cache().await; - if let Some(mut info) = group_cache.get(¬ification.group_jid).await { + if let Some(info) = group_cache.get(¬ification.group_jid).await { + let mut info = Arc::unwrap_or_clone(info); info.remove_participants(&users); group_cache - .insert(notification.group_jid.clone(), info) + .insert(notification.group_jid.clone(), Arc::new(info)) .await; debug!( target: "Client/Group", diff --git a/src/send.rs b/src/send.rs index b91797367..8815f4678 100644 --- a/src/send.rs +++ b/src/send.rs @@ -14,6 +14,26 @@ use wacore_binary::builder::NodeBuilder; use wacore_binary::{Jid, JidExt as _, Server}; use waproto::whatsapp as wa; +/// Returns a `GroupInfo` whose participant list is guaranteed to contain our own +/// sending JID, without deep-cloning the shared (cached) metadata in the common +/// case where the server's participant list already includes us. +fn ensure_self_in_group( + info: std::sync::Arc, + own_sending_jid: &Jid, +) -> std::sync::Arc { + if info + .participants + .iter() + .any(|participant| participant.is_same_user_as(own_sending_jid)) + { + info + } else { + let mut owned = (*info).clone(); + owned.participants.push(own_sending_jid.to_non_ad()); + std::sync::Arc::new(owned) + } +} + /// Options for [`Client::send_message_with_options`]. #[derive(Debug, Clone, Default)] pub struct SendOptions { @@ -481,6 +501,18 @@ impl Client { .map(|(_all, needs)| needs) }; + // prepare_group_stanza and ensure_status_participants both read the + // participant list and expect self present. Done after SKDM resolution + // to preserve the prior ordering (resolve ran without self appended). + let own_status_base = own_lid.to_non_ad(); + if !group_info + .participants + .iter() + .any(|participant| participant.is_same_user_as(&own_status_base)) + { + group_info.participants.push(own_status_base); + } + // `` describes the POSTER's privacy on their own // status. Reactions go through WA Web's addon path and never visit // `WAWebEncryptAndSendStatusMsg`; attaching the meta on a reaction @@ -499,7 +531,7 @@ impl Client { &*self.runtime, &mut stores, self, - &mut group_info, + &group_info, &own_jid, &own_lid, account_info.as_deref(), @@ -544,7 +576,7 @@ impl Client { &*self.runtime, &mut stores_retry, self, - &mut group_info, + &group_info, &own_jid, &own_lid, account_info.as_deref(), @@ -1080,7 +1112,7 @@ impl Client { } else if to.is_group() { // No send-level lock: encrypt_group_message serializes the // sender-key chain advance per (group, sender) at the cipher. - let mut group_info = self.groups().query_info(&to).await?; + let group_info = self.groups().query_info(&to).await?; let mut device_snapshot = self.persistence_manager.get_device_snapshot().await; let account_info = device_snapshot.account.take(); @@ -1104,13 +1136,9 @@ impl Client { crate::types::message::AddressingMode::Pn => (own_jid.clone(), "pn"), }; - if !group_info - .participants - .iter() - .any(|participant| participant.is_same_user_as(&own_sending_jid)) - { - group_info.participants.push(own_sending_jid.to_non_ad()); - } + // resolve_skdm_targets and prepare_group_stanza both read the + // participant list and expect self to be present. + let group_info = ensure_self_in_group(group_info, &own_sending_jid); let force_skdm = { use wacore::libsignal::store::sender_key_name::SenderKeyName; @@ -1182,7 +1210,7 @@ impl Client { &*self.runtime, &mut stores, self, - &mut group_info, + &group_info, &own_jid, &own_lid, account_info.as_deref(), @@ -1230,7 +1258,7 @@ impl Client { &*self.runtime, &mut stores_retry, self, - &mut group_info, + &group_info, &own_jid, &own_lid, account_info.as_deref(), @@ -2030,6 +2058,32 @@ mod tests { use super::*; use std::str::FromStr; + #[test] + fn ensure_self_in_group_shares_when_present_and_appends_when_absent() { + use wacore::client::context::GroupInfo; + use wacore::types::message::AddressingMode; + + let own: Jid = "999999999999@s.whatsapp.net".parse().unwrap(); + let other: Jid = "111111111111@s.whatsapp.net".parse().unwrap(); + + // Self already a member (the common case): the shared Arc passes through + // untouched, with no deep clone of the participant list. + let with_self = std::sync::Arc::new(GroupInfo::new( + vec![other.to_non_ad(), own.to_non_ad()], + AddressingMode::Pn, + )); + let out = ensure_self_in_group(with_self.clone(), &own); + assert!(std::sync::Arc::ptr_eq(&with_self, &out)); + + // Self missing: a fresh GroupInfo is built with self appended. + let without_self = + std::sync::Arc::new(GroupInfo::new(vec![other.to_non_ad()], AddressingMode::Pn)); + let out = ensure_self_in_group(without_self.clone(), &own); + assert!(!std::sync::Arc::ptr_eq(&without_self, &out)); + assert_eq!(out.participants.len(), 2); + assert!(out.participants.iter().any(|p| p.is_same_user_as(&own))); + } + #[tokio::test] async fn send_message_to_status_without_reaction_errors() { let client = crate::test_utils::create_test_client().await; diff --git a/wacore/benches/send_receive_benchmark.rs b/wacore/benches/send_receive_benchmark.rs index 06fb35fef..1ae14a2ec 100644 --- a/wacore/benches/send_receive_benchmark.rs +++ b/wacore/benches/send_receive_benchmark.rs @@ -14,6 +14,7 @@ use wacore::runtime::{AbortHandle, Runtime}; use wacore::send::{SignalStores, prepare_group_stanza, prepare_peer_stanza}; use wacore::types::jid::{JidExt, make_sender_key_name}; use wacore::types::message::AddressingMode; +use wacore_binary::JidExt as _; use wacore_binary::jid::Jid; use wacore_binary::marshal::marshal; use wacore_binary::node::{Node, NodeContent}; @@ -381,8 +382,14 @@ impl SendContextResolver for MockResolver { ) -> Result, anyhow::Error> { Ok(HashMap::new()) } - async fn resolve_group_info(&self, _: &Jid) -> Result { - Ok(GroupInfo::new(self.0.clone(), AddressingMode::Pn)) + async fn resolve_group_info( + &self, + _: &Jid, + ) -> Result, anyhow::Error> { + Ok(std::sync::Arc::new(GroupInfo::new( + self.0.clone(), + AddressingMode::Pn, + ))) } } @@ -638,8 +645,7 @@ fn setup_group_recv() -> GrpRecvData { // (server strips before forwarding to recipients) let resolver = MockResolver(vec![bob.jid.clone()]); let own_jid = alice.jid.clone(); - let mut group_info = - GroupInfo::new(vec![bob.jid.clone(), alice.jid.clone()], AddressingMode::Pn); + let group_info = GroupInfo::new(vec![bob.jid.clone(), alice.jid.clone()], AddressingMode::Pn); let mut stores = SignalStores { sender_key_store: &mut alice.sender_keys, @@ -654,7 +660,7 @@ fn setup_group_recv() -> GrpRecvData { &runtime, &mut stores, &resolver, - &mut group_info, + &group_info, &own_jid, &own_jid, None, @@ -719,6 +725,14 @@ fn run_group_send(d: &mut GrpSendData) { // itself and keeps None. let all_devices_for_phash = (!d.force_skdm).then(|| d.participants.clone()); let mut group_info = GroupInfo::new(std::mem::take(&mut d.participants), AddressingMode::Pn); + let own_base = own_jid.to_non_ad(); + if !group_info + .participants + .iter() + .any(|p| p.is_same_user_as(&own_base)) + { + group_info.participants.push(own_base); + } let mut stores = SignalStores { sender_key_store: &mut d.alice.sender_keys, session_store: &mut d.alice.sessions, @@ -731,7 +745,7 @@ fn run_group_send(d: &mut GrpSendData) { &d.runtime, &mut stores, &d.resolver, - &mut group_info, + &group_info, &own_jid, &own_jid, None, diff --git a/wacore/src/client/context.rs b/wacore/src/client/context.rs index e1c0458fd..31ee37e28 100644 --- a/wacore/src/client/context.rs +++ b/wacore/src/client/context.rs @@ -2,6 +2,7 @@ use crate::libsignal::protocol::PreKeyBundle; use crate::types::message::AddressingMode; use async_trait::async_trait; use std::collections::HashMap; +use std::sync::Arc; use wacore_binary::CompactString; use wacore_binary::Jid; @@ -153,7 +154,7 @@ pub trait SendContextResolver: crate::sync_marker::MaybeSendSync { jids: &[Jid], ) -> Result, anyhow::Error>; - async fn resolve_group_info(&self, jid: &Jid) -> Result; + async fn resolve_group_info(&self, jid: &Jid) -> Result, anyhow::Error>; /// Get the LID (Linked ID) for a phone number, if known. /// This is used to find existing sessions that were established under a LID address diff --git a/wacore/src/send.rs b/wacore/src/send.rs index 323959b5c..4dcb77423 100644 --- a/wacore/src/send.rs +++ b/wacore/src/send.rs @@ -1441,7 +1441,9 @@ pub async fn prepare_group_stanza< runtime: &dyn Runtime, stores: &mut SignalStores<'a, S, I, P, SP>, resolver: &dyn SendContextResolver, - group_info: &mut GroupInfo, + // Caller guarantees `own_base_jid` is already present in `participants`, so + // this reads the shared (Arc-backed) metadata without cloning it. + group_info: &GroupInfo, own_jid: &Jid, own_lid: &Jid, account: Option<&wa::AdvSignedDeviceIdentity>, @@ -1475,13 +1477,6 @@ pub async fn prepare_group_stanza< }; let own_base_jid = own_sending_jid.to_non_ad(); - if !group_info - .participants - .iter() - .any(|participant| participant.is_same_user_as(&own_base_jid)) - { - group_info.participants.push(own_base_jid.clone()); - } let mut message_children: Vec = Vec::new(); let mut includes_prekey_message = false; @@ -2433,7 +2428,7 @@ mod tests { Ok(result) } - async fn resolve_group_info(&self, _jid: &Jid) -> Result { + async fn resolve_group_info(&self, _jid: &Jid) -> Result> { unimplemented!("resolve_group_info not needed for send.rs tests") } @@ -4817,7 +4812,7 @@ mod tests { let resolver = MockSendContextResolver::new(); let rt = TokioTestRuntime; - let mut group_info = GroupInfo::new( + let group_info = GroupInfo::new( vec![own_jid.to_non_ad(), a.to_non_ad(), b.to_non_ad()], AddressingMode::Pn, ); @@ -4830,7 +4825,7 @@ mod tests { &rt, &mut stores, &resolver, - &mut group_info, + &group_info, &own_jid, &own_lid, None,