diff --git a/src/client/context_impl.rs b/src/client/context_impl.rs index 9d3b221f8..b04218a88 100644 --- a/src/client/context_impl.rs +++ b/src/client/context_impl.rs @@ -59,4 +59,22 @@ impl SendContextResolver for Client { fn on_local_identity_change(&self, jid: &Jid) { self.react_to_local_identity_change(jid); } + + async fn lock_device_sessions( + &self, + device_jids: &[Jid], + ) -> wacore::client::context::SessionLockGuard { + use wacore::client::context::SessionLockGuard; + if device_jids.is_empty() { + return SessionLockGuard::none(); + } + // Reuse the DM path's helpers so both lock the identical per-device mutexes. + let keys = self.build_session_lock_keys(device_jids).await; + let mutexes = self.session_mutexes_for(&keys).await; + let mut guards = Vec::with_capacity(mutexes.len()); + for mutex in &mutexes { + guards.push(mutex.lock_arc().await); + } + SessionLockGuard::hold(Box::new(guards)) + } } diff --git a/src/send/mod.rs b/src/send/mod.rs index 2a45395ff..3f3e8ebc5 100644 --- a/src/send/mod.rs +++ b/src/send/mod.rs @@ -2136,6 +2136,43 @@ mod tests { assert!(out.participants.iter().any(|p| p.is_same_user_as(&own))); } + // The group SKDM pairwise fan-out must hold the SAME per-device session mutex + // the DM path locks, so the two can't advance a shared device's ratchet at + // once. Acquiring the group lock must block the DM per-device lock. + #[tokio::test] + async fn group_skdm_lock_shares_dm_per_device_session_mutex() { + use wacore::client::context::SendContextResolver; + + let client = crate::test_utils::create_test_client().await; + let device: Jid = "15551234567:3@s.whatsapp.net".parse().unwrap(); + + // The exact mutex the DM send path would lock for this device. + let keys = client + .build_session_lock_keys(std::slice::from_ref(&device)) + .await; + let dm_mutexes = client.session_mutexes_for(&keys).await; + assert_eq!(dm_mutexes.len(), 1); + assert!( + dm_mutexes[0].try_lock().is_some(), + "uncontended before the group lock" + ); + + // Hold the group SKDM lock for the same device. + let guard = client + .lock_device_sessions(std::slice::from_ref(&device)) + .await; + assert!( + dm_mutexes[0].try_lock().is_none(), + "group SKDM fan-out must block the DM per-device session lock" + ); + + drop(guard); + assert!( + dm_mutexes[0].try_lock().is_some(), + "the per-device session lock releases when the group guard drops" + ); + } + #[tokio::test] async fn send_message_to_status_without_reaction_errors() { let client = crate::test_utils::create_test_client().await; diff --git a/wacore/src/client/context.rs b/wacore/src/client/context.rs index 9ae89505f..a7cb01a81 100644 --- a/wacore/src/client/context.rs +++ b/wacore/src/client/context.rs @@ -188,6 +188,28 @@ impl crate::stats::HeapSize for GroupInfo { } } +/// Opaque RAII holder for the per-device pairwise session locks a +/// [`SendContextResolver`] acquires around the group SKDM fan-out. Wacore holds it +/// across the fan-out and drops it to release; the concrete guard type lives in the +/// platform crate, since the per-address lock cache is not part of the portable core. +#[must_use = "the session locks release the moment this guard is dropped"] +pub struct SessionLockGuard( + // Held purely for its `Drop` (releases the locks); never read. + #[allow(dead_code)] Option>, +); + +impl SessionLockGuard { + /// No locks held — the default resolver behavior (tests/benches don't race). + pub fn none() -> Self { + Self(None) + } + + /// Hold `guards` until this value is dropped. + pub fn hold(guards: Box) -> Self { + Self(Some(guards)) + } +} + #[cfg_attr(target_arch = "wasm32", async_trait(?Send))] #[cfg_attr(not(target_arch = "wasm32"), async_trait)] pub trait SendContextResolver: crate::sync_marker::MaybeSendSync { @@ -225,6 +247,17 @@ pub trait SendContextResolver: crate::sync_marker::MaybeSendSync { fn on_local_identity_change(&self, jid: &Jid) { let _ = jid; } + + /// Acquire the per-device pairwise session locks for the SKDM fan-out targets, + /// in the same deadlock-free order the DM send path uses, so a group send and a + /// concurrent DM (or another group send) sharing a device can't advance that + /// device's pairwise ratchet at once and drop a chain step. The `sender_key_lock` + /// only serializes the sender-key chain, not these pairwise sessions. Default: + /// no-op — tests and benches don't race concurrent sends. + async fn lock_device_sessions(&self, device_jids: &[Jid]) -> SessionLockGuard { + let _ = device_jids; + SessionLockGuard::none() + } } #[cfg(test)] diff --git a/wacore/src/send/group.rs b/wacore/src/send/group.rs index e15430051..3d632a6bb 100644 --- a/wacore/src/send/group.rs +++ b/wacore/src/send/group.rs @@ -341,6 +341,16 @@ pub async fn prepare_group_stanza( let sender_key_name = make_sender_key_name(&to_jid, &own_sending_jid.to_protocol_address()); + // Hold the per-device session locks the DM path uses across BOTH the X3DH setup + // and the SKDM fan-out below, so a concurrent DM or group send sharing a device + // can't race that device's pairwise session (create or ratchet-advance). The DM + // path holds the same locks across all of prepare_dm_stanza; sender_key_lock only + // serializes the sender-key chain. Acquired before sender_key_lock so the whole + // send path takes session -> sender-key order (no other path takes the reverse). + let session_guard = resolver + .lock_device_sessions(distribution_list.as_deref().unwrap_or(&[])) + .await; + // Establish missing pairwise sessions (prekey fetch + X3DH) for the SKDM // targets before taking the chain lock, so the chain critical section // below never spans a network RTT — concurrent sends to the same group @@ -385,11 +395,9 @@ pub async fn prepare_group_stanza( // The lock spans SKDM creation, the pairwise SKDM fan-out, and the skmsg // encrypt. Creating the SKDM snapshots the sender key and the skmsg uses it, - // so those must be atomic. The fan-out also mutates the shared per-device - // Signal sessions, which the group path (unlike the DM path) does not lock, - // so it has to stay serialized here too or concurrent same-group sends race - // those sessions. Dropped after the encrypt so only the stanza build runs - // off the serialization point. + // so those must be atomic. Per-device pairwise sessions are guarded separately + // by `session_guard` above. Dropped after the encrypt so only the stanza build + // runs off the serialization point. let chain_lock = stores .sender_key_store .sender_key_lock(&sender_key_name) @@ -476,6 +484,11 @@ pub async fn prepare_group_stanza( } } + // The skmsg encrypt only advances the sender-key chain, not any pairwise + // session, so release the per-device locks now instead of holding them across + // it (avoids head-of-line blocking a concurrent DM to a shared device). + drop(session_guard); + let skmsg = encrypt_group_message( stores.sender_key_store, &sender_key_name,