diff --git a/src/features/signal.rs b/src/features/signal.rs index 469f62eae..2220a0ec5 100644 --- a/src/features/signal.rs +++ b/src/features/signal.rs @@ -265,12 +265,7 @@ impl<'a> Signal<'a> { // Acquire per-device session locks before encrypting (matches DM send path) let lock_jids = self.client.build_session_lock_keys(&device_jids).await; - let mut session_mutexes = Vec::with_capacity(lock_jids.len()); - let mut lock_buf = String::with_capacity(64); - for jid in &lock_jids { - wacore::types::jid::write_protocol_address_to(jid, &mut lock_buf); - session_mutexes.push(self.client.session_lock_for(&lock_buf).await); - } + let session_mutexes = self.client.session_mutexes_for(&lock_jids).await; let mut _session_guards = Vec::with_capacity(session_mutexes.len()); for mutex in &session_mutexes { _session_guards.push(mutex.lock().await); diff --git a/src/send.rs b/src/send.rs index 9271bd1c7..0e249ab97 100644 --- a/src/send.rs +++ b/src/send.rs @@ -1078,13 +1078,7 @@ impl Client { } let lock_jids = self.build_session_lock_keys(&all_dm_jids).await; - - let mut _session_mutexes = Vec::with_capacity(lock_jids.len()); - let mut lock_buf = String::with_capacity(64); - for jid in &lock_jids { - wacore::types::jid::write_protocol_address_to(jid, &mut lock_buf); - _session_mutexes.push(self.session_lock_for(&lock_buf).await); - } + let _session_mutexes = self.session_mutexes_for(&lock_jids).await; let mut _session_guards = Vec::with_capacity(_session_mutexes.len()); for mutex in &_session_mutexes { _session_guards.push(mutex.lock().await); @@ -1487,10 +1481,7 @@ impl Client { /// Build sorted, deduplicated per-device session lock keys. /// INVARIANT: Keys are sorted to prevent deadlocks when acquiring multiple /// session locks (e.g. DM sends that encrypt for recipient + own devices). - /// Keys match the decrypt path format so send and receive serialize correctly. - /// Returns resolved encryption JIDs sorted in a consistent order for - /// deadlock-free lock acquisition. Sorts by Jid fields directly — no - /// intermediate ProtocolAddress/String allocations for sorting. + /// Resolve encryption JIDs and sort for deadlock-free lock acquisition. pub(crate) async fn build_session_lock_keys(&self, device_jids: &[Jid]) -> Vec { let mut keys: Vec = Vec::with_capacity(device_jids.len()); for jid in device_jids { @@ -1501,6 +1492,20 @@ impl Client { keys } + /// Fetch per-device session mutexes in deadlock-free order. + pub(crate) async fn session_mutexes_for( + &self, + jids: &[Jid], + ) -> Vec>> { + let mut mutexes = Vec::with_capacity(jids.len()); + let mut buf = String::with_capacity(64); + for jid in jids { + wacore::types::jid::write_protocol_address_to(jid, &mut buf); + mutexes.push(self.session_lock_for(&buf).await); + } + mutexes + } + /// Build tctoken timing config from AB props, falling back to defaults. pub(crate) async fn tc_token_config(&self) -> wacore::iq::tctoken::TcTokenConfig { use wacore::iq::props::config_codes; diff --git a/wacore/src/send.rs b/wacore/src/send.rs index 5dfe95846..ad3e6eff7 100644 --- a/wacore/src/send.rs +++ b/wacore/src/send.rs @@ -1038,8 +1038,7 @@ pub async fn prepare_group_stanza< jids_to_resolve.push(own_jid_to_check); } - let mut seen_users = HashSet::new(); - jids_to_resolve.retain(|jid| seen_users.insert((jid.user.clone(), jid.server.clone()))); + crate::types::jid::sort_dedup_by_user(&mut jids_to_resolve); log::debug!( "Resolving devices for {} participants", @@ -1067,10 +1066,7 @@ pub async fn prepare_group_stanza< // both convert to 100000037037034:33@lid). // Key on (user, server, agent, device) — excludes `integrator` which is not // part of the wire JID identity used in and phash. - let mut seen = HashSet::new(); - resolved_list.retain(|jid| { - seen.insert((jid.user.clone(), jid.server.clone(), jid.agent, jid.device)) - }); + crate::types::jid::sort_dedup_by_device(&mut resolved_list); // Filter devices for SKDM distribution: // - Exclude the exact sending device (own_sending_jid) - we already have our own sender key @@ -1211,7 +1207,7 @@ pub async fn prepare_group_stanza< let stanza_type = stanza_type_from_message(message); let mut stanza_builder = NodeBuilder::new("message") - .attr("to", to_jid.clone()) + .attr("to", to_jid) .attr("id", request_id) .attr("type", stanza_type); diff --git a/wacore/src/types/jid.rs b/wacore/src/types/jid.rs index b0e60e49e..bc6711a9c 100644 --- a/wacore/src/types/jid.rs +++ b/wacore/src/types/jid.rs @@ -7,8 +7,7 @@ fn mapped_server(s: &str) -> &str { if s == "s.whatsapp.net" { "c.us" } else { s } } -/// Write the protocol address lock key (`{user}[:{device}]@{server}.0`) -/// into `buf`, reusing its allocation. Zero heap allocations. +/// Write the protocol address lock key into `buf`, reusing its allocation. pub fn write_protocol_address_to(jid: &Jid, buf: &mut String) { use std::fmt::Write; buf.clear(); @@ -24,7 +23,6 @@ pub fn write_protocol_address_to(jid: &Jid, buf: &mut String) { } /// Consistent ordering for deadlock-free multi-lock acquisition. -/// Compares Jid fields directly — no String allocation needed. pub fn cmp_for_lock_order(a: &Jid, b: &Jid) -> std::cmp::Ordering { mapped_server(&a.server) .cmp(mapped_server(&b.server)) @@ -32,6 +30,26 @@ pub fn cmp_for_lock_order(a: &Jid, b: &Jid) -> std::cmp::Ordering { .then_with(|| a.device.cmp(&b.device)) } +/// Sort and deduplicate by user identity (user + server). +pub fn sort_dedup_by_user(jids: &mut Vec) { + jids.sort_unstable_by(|a, b| a.user.cmp(&b.user).then_with(|| a.server.cmp(&b.server))); + jids.dedup_by(|a, b| a.user == b.user && a.server == b.server); +} + +/// Sort and deduplicate by device identity (user + server + agent + device). +pub fn sort_dedup_by_device(jids: &mut Vec) { + jids.sort_unstable_by(|a, b| { + a.user + .cmp(&b.user) + .then_with(|| a.server.cmp(&b.server)) + .then_with(|| a.agent.cmp(&b.agent)) + .then_with(|| a.device.cmp(&b.device)) + }); + jids.dedup_by(|a, b| { + a.user == b.user && a.server == b.server && a.agent == b.agent && a.device == b.device + }); +} + pub trait JidExt { fn to_protocol_address(&self) -> ProtocolAddress;