fix(send): lock per-device sessions across the group SKDM fan-out

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))
+    }
 }

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;

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<Box<dyn crate::sync_marker::MaybeSendSync>>,
+);
+
+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<dyn crate::sync_marker::MaybeSendSync>) -> 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)]

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,