fix(signal): serialize sender-key mutations

src/cache_config.rs

@@ -204,10 +204,9 @@ pub struct CacheConfig {
     pub session_locks_capacity: u64,
     /// Per-chat lane capacity (combined lock + queue). Default: 5000.
     pub chat_lanes_capacity: u64,
-    /// Per-group cold sender-key distribution lock capacity. Default: 512
-    /// (far above any realistic number of groups distributing at once; an
-    /// evicted live lock only lets one extra send repeat that group's
-    /// fan-out, the ratchet stays correct under the chain lock).
+    /// Per-group cold sender-key distribution lock capacity. Default: 512.
+    /// Soft cap: a live lane is never evicted, so the map may briefly exceed
+    /// this under concurrent fan-out instead of breaking tracker ordering.
     pub group_distribution_locks_capacity: u64,
     /// Per-chat resend rate-limiter capacity: one token-bucket entry per group
     /// recently driving retry resends. Keep above the count of concurrently

src/client.rs

@@ -225,6 +225,11 @@ pub struct MemoryReport {
     // -- Capacity-only caches (coordination, counts only) --
     pub session_locks: u64,
     pub chat_lanes: u64,
+    pub group_distribution_locks: u64,
+    /// Cumulative capacity evictions; poll successive reports to derive a rate.
+    pub group_distribution_lock_evictions: u64,
+    /// Cumulative attempts that kept a live lane and temporarily exceeded capacity.
+    pub group_distribution_lock_eviction_blocks: u64,
     pub resend_rate_limiter_chats: u64,
     // -- Unbounded collections --
     pub response_waiters: usize,
@@ -295,6 +300,13 @@ impl std::fmt::Display for MemoryReport {
         writeln!(f, "--- Capacity-only caches ---")?;
         writeln!(f, "  session_locks:          {}", self.session_locks)?;
         writeln!(f, "  chat_lanes:             {}", self.chat_lanes)?;
+        writeln!(
+            f,
+            "  group_dist_locks:       {} (evicted: {}, blocked: {})",
+            self.group_distribution_locks,
+            self.group_distribution_lock_evictions,
+            self.group_distribution_lock_eviction_blocks
+        )?;
         writeln!(
             f,
             "  resend_rl_chats:        {}",

src/client/accessors.rs

@@ -156,6 +156,7 @@ impl Client {
             .group_devices_memo
             .memory_stats(|k, v| k.heap_bytes() + v.heap_bytes())
             .await;
+        let group_distribution_locks = self.group_distribution_locks.capacity_stats().await;
 
         // Each count read into a local so no two guards are ever held at once.
         let response_waiters = self.response_waiters_guard().len();
@@ -178,6 +179,9 @@ impl Client {
             pdo_requested: self.pdo_requested.entry_count(),
             session_locks: self.session_locks.entry_count(),
             chat_lanes: self.chat_lanes.entry_count(),
+            group_distribution_locks: group_distribution_locks.entries,
+            group_distribution_lock_evictions: group_distribution_locks.evictions,
+            group_distribution_lock_eviction_blocks: group_distribution_locks.eviction_blocks,
             resend_rate_limiter_chats: self.resend_rate_limiter.entry_count(),
             response_waiters,
             node_waiters: self.node_waiter_count.load(Ordering::Relaxed),

src/client/device_registry.rs

@@ -1032,7 +1032,7 @@ impl Client {
 mod tests {
     use super::*;
     use crate::lid_pn_cache::LearningSource;
-    use crate::test_utils::create_test_client_with_failing_http;
+    use crate::test_utils::{create_test_client_with_failing_http, wait_for_lock_waiter};
     use std::sync::Arc;
 
     async fn create_test_client() -> Arc<Client> {
@@ -2700,7 +2700,8 @@ mod tests {
         use wacore::types::jid::JidExt;
 
         let client = create_test_client().await;
-        let group = "120363000000000001@g.us";
+        let group: Jid = "120363000000000001@g.us".parse().unwrap();
+        let group_id = group.to_string();
         let own_lid = Jid::from_str("193832511623409:13@lid").unwrap();
         client
             .persistence_manager
@@ -2709,7 +2710,7 @@ mod tests {
             )))
             .await;
 
-        let sk_name = SenderKeyName::from_parts(group, own_lid.to_protocol_address().as_str());
+        let sk_name = SenderKeyName::from_parts(&group_id, own_lid.to_protocol_address().as_str());
         client
             .signal_cache
             .put_sender_key(&sk_name, SenderKeyRecord::new_empty())
@@ -2718,7 +2719,7 @@ mod tests {
         client
             .persistence_manager
             .set_sender_key_status(
-                group,
+                &group_id,
                 &[
                     ("271060335329480:0@lid", true),
                     ("77610646245392:0@lid", true),
@@ -2728,7 +2729,7 @@ mod tests {
             .unwrap();
 
         client
-            .rotate_sender_key_on_participant_remove(group, &["271060335329480"])
+            .rotate_sender_key_on_participant_remove(&group, &["271060335329480"])
             .await;
 
         let device_snapshot = client.persistence_manager.get_device_snapshot();
@@ -2744,7 +2745,7 @@ mod tests {
 
         let rows = client
             .persistence_manager
-            .get_sender_key_devices(group)
+            .get_sender_key_devices(&group_id)
             .await
             .unwrap();
         assert!(rows.is_empty(), "sender_key_devices must be cleared");
@@ -2760,7 +2761,8 @@ mod tests {
         use wacore::types::jid::JidExt;
 
         let client = create_test_client().await;
-        let group = "120363000000000001@g.us";
+        let group: Jid = "120363000000000001@g.us".parse().unwrap();
+        let group_id = group.to_string();
         let own_lid = Jid::from_str("193832511623409:13@lid").unwrap();
         client
             .persistence_manager
@@ -2769,20 +2771,20 @@ mod tests {
             )))
             .await;
 
-        let sk_name = SenderKeyName::from_parts(group, own_lid.to_protocol_address().as_str());
+        let sk_name = SenderKeyName::from_parts(&group_id, own_lid.to_protocol_address().as_str());
         client
             .signal_cache
             .put_sender_key(&sk_name, SenderKeyRecord::new_empty())
             .await;
 
         client
             .persistence_manager
-            .set_sender_key_status(group, &[("271060335329480:0@lid", false)])
+            .set_sender_key_status(&group_id, &[("271060335329480:0@lid", false)])
             .await
             .unwrap();
 
         client
-            .rotate_sender_key_on_participant_remove(group, &["271060335329480"])
+            .rotate_sender_key_on_participant_remove(&group, &["271060335329480"])
             .await;
 
         let device_snapshot = client.persistence_manager.get_device_snapshot();
@@ -2796,4 +2798,186 @@ mod tests {
             "sender key must survive when removed had no key"
         );
     }
+
+    #[tokio::test]
+    async fn rotation_waits_for_in_flight_sender_key_advance() {
+        use wacore::libsignal::protocol::{
+            KeyPair, SENDERKEY_MESSAGE_CURRENT_VERSION, SenderKeyRecord, group_encrypt,
+        };
+        use wacore::libsignal::store::sender_key_name::SenderKeyName;
+        use wacore::types::jid::JidExt;
+
+        let client = create_test_client().await;
+        let group: Jid = "120363000000000003@g.us".parse().unwrap();
+        let group_id = group.to_string();
+        let own_lid: Jid = "193832511623410:13@lid".parse().unwrap();
+        client
+            .persistence_manager
+            .process_command(crate::store::commands::DeviceCommand::SetLid(Some(
+                own_lid.clone(),
+            )))
+            .await;
+
+        let name = SenderKeyName::from_parts(&group_id, own_lid.to_protocol_address().as_str());
+        let mut rng = rand::make_rng::<rand::rngs::StdRng>();
+        let key_pair = KeyPair::generate(&mut rng);
+        let mut record = SenderKeyRecord::new_empty();
+        record
+            .add_sender_key_state(
+                SENDERKEY_MESSAGE_CURRENT_VERSION,
+                7,
+                0,
+                &[9; 32],
+                key_pair.public_key,
+                Some(key_pair.private_key),
+            )
+            .unwrap();
+        client.signal_cache.put_sender_key(&name, record).await;
+
+        let chain_lock = client.signal_cache.sender_key_lock(&name).await;
+        let held = chain_lock.lock().await;
+        let lock_refs = Arc::strong_count(&chain_lock);
+        let started = Arc::new(tokio::sync::Barrier::new(2));
+        let rotation = tokio::spawn({
+            let client = client.clone();
+            let group = group.clone();
+            let started = started.clone();
+            async move {
+                started.wait().await;
+                client.force_rotate_own_sender_key(&group).await;
+            }
+        });
+
+        started.wait().await;
+        wait_for_lock_waiter(&chain_lock, lock_refs).await;
+        let snapshot = client.persistence_manager.get_device_snapshot();
+        assert!(
+            client
+                .signal_cache
+                .get_sender_key(&name, &*snapshot.backend)
+                .await
+                .unwrap()
+                .is_some(),
+            "rotation must wait for the in-flight advance"
+        );
+
+        let mut sender_key_store = client.sender_key_adapter().await;
+        group_encrypt(
+            &mut sender_key_store,
+            &name,
+            b"in-flight ciphertext",
+            &mut rng,
+        )
+        .await
+        .expect("advance under the held chain lock");
+        drop(held);
+
+        tokio::time::timeout(std::time::Duration::from_secs(5), rotation)
+            .await
+            .expect("rotation must resume")
+            .expect("rotation task");
+        assert!(
+            client
+                .signal_cache
+                .get_sender_key(&name, &*snapshot.backend)
+                .await
+                .unwrap()
+                .is_none(),
+            "rotation must retire the state written by the in-flight advance"
+        );
+    }
+
+    #[tokio::test]
+    async fn participant_rotation_audit_waits_for_group_distribution_guard() {
+        use wacore::libsignal::protocol::SenderKeyRecord;
+        use wacore::libsignal::store::sender_key_name::SenderKeyName;
+        use wacore::types::jid::JidExt;
+
+        let client = create_test_client().await;
+        let group: Jid = "120363000000000004@g.us".parse().unwrap();
+        let group_id = group.to_string();
+        let own_lid: Jid = "193832511623411:13@lid".parse().unwrap();
+        client
+            .persistence_manager
+            .process_command(crate::store::commands::DeviceCommand::SetLid(Some(
+                own_lid.clone(),
+            )))
+            .await;
+
+        let name = SenderKeyName::from_parts(&group_id, own_lid.to_protocol_address().as_str());
+        client
+            .signal_cache
+            .put_sender_key(&name, SenderKeyRecord::new_empty())
+            .await;
+        client
+            .persistence_manager
+            .set_sender_key_status(&group_id, &[("271060335329481:0@lid", true)])
+            .await
+            .unwrap();
+
+        let held = client.group_distribution_lock(&group).await;
+        let lock = client
+            .group_distribution_locks
+            .get(&group)
+            .await
+            .expect("cached distribution lock");
+        let lock_refs = Arc::strong_count(&lock);
+        let started = Arc::new(tokio::sync::Barrier::new(2));
+        let rotation = tokio::spawn({
+            let client = client.clone();
+            let group = group.clone();
+            let started = started.clone();
+            async move {
+                started.wait().await;
+                client
+                    .rotate_sender_key_on_participant_remove(&group, &["271060335329481"])
+                    .await;
+            }
+        });
+
+        started.wait().await;
+        wait_for_lock_waiter(&lock, lock_refs).await;
+        let snapshot = client.persistence_manager.get_device_snapshot();
+        assert!(
+            client
+                .signal_cache
+                .get_sender_key(&name, &*snapshot.backend)
+                .await
+                .unwrap()
+                .is_some(),
+            "rotation must not delete before the active distribution ends"
+        );
+        assert_eq!(
+            client
+                .persistence_manager
+                .get_sender_key_devices(&group_id)
+                .await
+                .unwrap()
+                .len(),
+            1,
+            "rotation must not clear tracking before it owns the distribution lane"
+        );
+
+        drop(held);
+        tokio::time::timeout(std::time::Duration::from_secs(5), rotation)
+            .await
+            .expect("rotation must resume")
+            .expect("rotation task");
+        assert!(
+            client
+                .signal_cache
+                .get_sender_key(&name, &*snapshot.backend)
+                .await
+                .unwrap()
+                .is_none()
+        );
+        assert!(
+            client
+                .persistence_manager
+                .get_sender_key_devices(&group_id)
+                .await
+                .unwrap()
+                .is_empty()
+        );
+    }
 }

src/client/lifecycle.rs

@@ -267,12 +267,10 @@ impl Client {
             group_devices_memo: Cache::builder()
                 .max_capacity(GROUP_DEVICES_MEMO_CAPACITY)
                 .build(),
-            // Evicting a lock whose guard is still held only lets one extra
-            // send re-run that group's fan-out (the pre-single-flight
-            // behavior); the sender-key chain lock still guarantees ratchet
-            // correctness.
+            // A live lane also protects recipient-tracker reset/update ordering.
             group_distribution_locks: Cache::builder()
                 .max_capacity(cache_config.group_distribution_locks_capacity.max(1))
+                .evict_guard(|m| Arc::strong_count(m) <= 1)
                 .build(),
             skdm_warm_memo: Cache::builder()
                 .max_capacity(GROUP_DEVICES_MEMO_CAPACITY)