fix(send): close SKDM flow gaps for forward secrecy and cache hygiene

src/appstate_sync.rs

@@ -164,6 +164,9 @@ mod tests {
         async fn clear_all_sender_key_devices(&self) -> StoreResult<()> {
             Ok(())
         }
+        async fn delete_sender_key_device_rows(&self, _: &[&str]) -> StoreResult<()> {
+            Ok(())
+        }
         async fn get_lid_mapping(&self, _: &str) -> StoreResult<Option<LidPnMappingEntry>> {
             Ok(None)
         }

src/client/sender_keys.rs

@@ -67,6 +67,72 @@ impl Client {
         Ok(())
     }
 
+    /// Forward-secrecy rotation when participants leave a group. Mirrors WA
+    /// Web's `removeParticipantInfo` (`GroupParticipantHelpers.js`): if any
+    /// removed user had `has_key=true`, delete the bot's own sender key for
+    /// the group and wipe `sender_key_devices` so the next send takes the
+    /// `force_skdm=true` path (`!key_exists`) and redistributes to all
+    /// remaining participants.
+    pub(crate) async fn rotate_sender_key_on_participant_remove(
+        &self,
+        group_jid: &str,
+        removed_user_ids: &[&str],
+    ) {
+        if removed_user_ids.is_empty() {
+            return;
+        }
+
+        // Read failure → rotate anyway. Better to pay the redistribute cost
+        // than leave the sender key in place after a removal we couldn't audit.
+        let (rows, read_failed) = match self
+            .persistence_manager
+            .get_sender_key_devices(group_jid)
+            .await
+        {
+            Ok(r) => (r, false),
+            Err(e) => {
+                log::warn!(
+                    "rotate_sender_key_on_participant_remove: read failed for {group_jid}: {e} \
+                     — rotating conservatively"
+                );
+                (Vec::new(), true)
+            }
+        };
+
+        let any_had_key = rows.iter().any(|(jid_str, has_key)| {
+            *has_key
+                && jid_str
+                    .parse::<Jid>()
+                    .ok()
+                    .is_some_and(|jid| removed_user_ids.iter().any(|u| *u == jid.user.as_str()))
+        });
+        if !read_failed && !any_had_key {
+            return;
+        }
+
+        use wacore::libsignal::store::sender_key_name::SenderKeyName;
+        use wacore::types::jid::JidExt;
+        let snapshot = self.persistence_manager.get_device_snapshot().await;
+        for own_jid in snapshot.lid.iter().chain(snapshot.pn.iter()) {
+            let sk_name =
+                SenderKeyName::from_parts(group_jid, own_jid.to_protocol_address().as_str());
+            self.signal_cache
+                .delete_sender_key(sk_name.cache_key())
+                .await;
+        }
+        self.flush_signal_cache_logged("rotate_sender_key_on_participant_remove", None)
+            .await;
+
+        if let Err(e) = self
+            .persistence_manager
+            .clear_sender_key_devices(group_jid)
+            .await
+        {
+            log::warn!("rotate_sender_key_on_participant_remove: clear DB failed: {e}");
+        }
+        self.sender_key_device_cache.invalidate(group_jid).await;
+    }
+
     /// Take a sent message for retry handling. Checks L1 cache first (if enabled),
     /// then falls back to DB. On miss, tries an alternate PN/LID key to handle
     /// mapping changes between send time and retry time (WAWebLidMigrationUtils

src/features/groups.rs

@@ -390,6 +390,9 @@ impl<'a> Groups<'a> {
                 info.remove_participants(&accepted);
                 group_cache.insert(jid.clone(), info).await;
             }
+            self.client
+                .rotate_sender_key_on_participant_remove(&jid.to_string(), &accepted)
+                .await;
         }
         Ok(result)
     }

src/handlers/notification.rs

@@ -1192,10 +1192,9 @@ async fn handle_group_notification(client: &Arc<Client>, node: Arc<OwnedNodeRef>
                 }
             }
             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(&notification.group_jid).await {
-                    let users: Vec<&str> =
-                        participants.iter().map(|p| p.jid.user.as_str()).collect();
                     info.remove_participants(&users);
                     group_cache
                         .insert(notification.group_jid.clone(), info)
@@ -1206,6 +1205,12 @@ async fn handle_group_notification(client: &Arc<Client>, node: Arc<OwnedNodeRef>
                         notification.group_jid, participants.len()
                     );
                 }
+                client
+                    .rotate_sender_key_on_participant_remove(
+                        &notification.group_jid.to_string(),
+                        &users,
+                    )
+                    .await;
             }
             _ => {}
         }

src/send.rs

@@ -696,10 +696,38 @@ impl Client {
                             client.invalidate_device_cache(&own_pn.user).await;
                         }
                     }
-                    client
-                        .sender_key_device_cache
-                        .invalidate(&jid.to_string())
-                        .await;
+                    let jid_str = jid.to_string();
+                    // Cache-only invalidation re-reads the same stale rows on
+                    // the next send. Drop the persisted state too so the next
+                    // send takes the full-distribution path. If the clear
+                    // fails, fall back to deleting the bot's own sender key
+                    // for the chat — the next send will see `!key_exists`
+                    // and force_skdm without depending on the tracker.
+                    if jid.is_group() || jid.is_status_broadcast() {
+                        let cleared = client
+                            .persistence_manager
+                            .clear_sender_key_devices(&jid_str)
+                            .await;
+                        if let Err(e) = cleared {
+                            log::warn!(
+                                "phash mismatch: clear_sender_key_devices failed: {e} — \
+                                 deleting own sender key as fallback to force redistribution"
+                            );
+                            use wacore::libsignal::store::sender_key_name::SenderKeyName;
+                            use wacore::types::jid::JidExt;
+                            let snapshot =
+                                client.persistence_manager.get_device_snapshot().await;
+                            for own in snapshot.lid.iter().chain(snapshot.pn.iter()) {
+                                let sk =
+                                    SenderKeyName::from_parts(&jid_str, own.to_protocol_address().as_str());
+                                client.signal_cache.delete_sender_key(sk.cache_key()).await;
+                            }
+                            let _ = client
+                                .flush_signal_cache_logged("phash-mismatch-fallback", None)
+                                .await;
+                        }
+                    }
+                    client.sender_key_device_cache.invalidate(&jid_str).await;
                     if invalidate_group_cache {
                         client.get_group_cache().await.invalidate(&jid).await;
                     }
@@ -966,13 +994,40 @@ impl Client {
                 let sender_key_name = SenderKeyName::from_parts(&to_str, sender_address.as_str());
 
                 let device_guard = device_store_arc.read().await;
-                let key_exists = self
+                let record = self
                     .signal_cache
                     .get_sender_key(&sender_key_name, &*device_guard.backend)
-                    .await?
-                    .is_some();
+                    .await?;
+                let key_exists = record.is_some();
+
+                // WA Web posts SenderKeyExpired with `PERIODIC_ROTATION` after
+                // a chain advances past a threshold. Captured-js doesn't show
+                // the value; 1000 mirrors common Signal hygiene defaults.
+                const SENDER_KEY_ROTATION_THRESHOLD: u32 = 1000;
+                let needs_rotation = record
+                    .and_then(|mut r| r.sender_key_state_mut().ok().cloned())
+                    .and_then(|state| state.sender_chain_key().map(|ck| ck.iteration()))
+                    .is_some_and(|iter| iter >= SENDER_KEY_ROTATION_THRESHOLD);
+                drop(device_guard);
+
+                if needs_rotation {
+                    log::info!(
+                        "Periodic sender-key rotation for {to} (chain iteration ≥ {SENDER_KEY_ROTATION_THRESHOLD})"
+                    );
+                    self.signal_cache
+                        .delete_sender_key(sender_key_name.cache_key())
+                        .await;
+                    if let Err(e) = self
+                        .persistence_manager
+                        .clear_sender_key_devices(&to_str)
+                        .await
+                    {
+                        log::warn!("periodic rotation: clear_sender_key_devices failed: {e}");
+                    }
+                    self.sender_key_device_cache.invalidate(&to_str).await;
+                }
 
-                force_key_distribution || !key_exists
+                force_key_distribution || !key_exists || needs_rotation
             };
 
             let mut store_adapter = self.signal_adapter_from(device_store_arc.clone());
@@ -1231,7 +1286,16 @@ impl Client {
         }
 
         if let Some((rx, phash, msg_id)) = ack {
-            self.spawn_phash_validation(rx, phash, tc_issue_target.clone(), false, msg_id);
+            // Group sends also invalidate group cache on mismatch — server's
+            // participant set diverged, the next send needs a fresh query.
+            let invalidate_group = tc_issue_target.is_group();
+            self.spawn_phash_validation(
+                rx,
+                phash,
+                tc_issue_target.clone(),
+                invalidate_group,
+                msg_id,
+            );
         }
 
         if let Some(update) = skdm_update {

src/sender_key_device_cache.rs

@@ -84,10 +84,23 @@ impl SenderKeyDeviceCache {
         self.inner.invalidate(group_jid).await;
     }
 
-    /// Invalidate all entries. Used on raw_id mismatch (identity change)
-    /// to force SKDM redistribution for all groups.
-    pub(crate) fn invalidate_all(&self) {
-        self.inner.invalidate_all();
+    /// Drop cache entries whose map indexes the given (user, device_id). Needed
+    /// after a device is removed: a future re-add of the same device_id would
+    /// otherwise hit a stale `has_key=true` entry and skip SKDM redistribution.
+    pub(crate) async fn invalidate_entries_for_device(&self, user: &str, device_id: u16) {
+        let to_drop: Vec<String> = self
+            .inner
+            .iter()
+            .filter_map(|(group_jid, map)| {
+                map.devices
+                    .get(user)
+                    .and_then(|devmap| devmap.get(&device_id))
+                    .map(|_| group_jid.as_ref().clone())
+            })
+            .collect();
+        for g in to_drop {
+            self.inner.invalidate(&g).await;
+        }
     }
 
     #[cfg(feature = "debug-diagnostics")]

src/store/persistence_manager.rs

@@ -253,6 +253,15 @@ impl PersistenceManager {
     pub async fn clear_sender_key_devices(&self, group_jid: &str) -> Result<(), StoreError> {
         self.backend.clear_sender_key_devices(group_jid).await
     }
+
+    pub async fn delete_sender_key_device_rows(
+        &self,
+        device_jids: &[&str],
+    ) -> Result<(), StoreError> {
+        self.backend
+            .delete_sender_key_device_rows(device_jids)
+            .await
+    }
 }
 
 #[cfg(test)]

storages/sqlite-storage/src/sqlite_store.rs

@@ -1854,6 +1854,30 @@ impl ProtocolStore for SqliteStore {
         .await
     }
 
+    async fn delete_sender_key_device_rows(&self, device_jids: &[&str]) -> Result<()> {
+        if device_jids.is_empty() {
+            return Ok(());
+        }
+        let device_id = self.device_id;
+        let owned: Arc<Vec<String>> = Arc::new(device_jids.iter().map(|s| s.to_string()).collect());
+        self.with_retry("delete_sender_key_device_rows", || {
+            let owned = Arc::clone(&owned);
+            Box::new(move |conn: &mut SqliteConnection| {
+                const CHUNK: usize = 190;
+                for chunk in owned.chunks(CHUNK) {
+                    diesel::delete(
+                        sender_key_devices::table
+                            .filter(sender_key_devices::device_jid.eq_any(chunk))
+                            .filter(sender_key_devices::device_id.eq(device_id)),
+                    )
+                    .execute(conn)?;
+                }
+                Ok(())
+            })
+        })
+        .await
+    }
+
     async fn get_lid_mapping(&self, lid: &str) -> Result<Option<LidPnMappingEntry>> {
         let pool = self.pool.clone();
         let device_id = self.device_id;

wacore/src/store/in_memory.rs

@@ -385,6 +385,18 @@ impl ProtocolStore for InMemoryBackend {
         Ok(())
     }
 
+    async fn delete_sender_key_device_rows(&self, device_jids: &[&str]) -> Result<()> {
+        if device_jids.is_empty() {
+            return Ok(());
+        }
+        let mut state = self.state.lock().await;
+        let targets: std::collections::HashSet<&str> = device_jids.iter().copied().collect();
+        for group_map in state.sender_key_devices.values_mut() {
+            group_map.retain(|jid, _| !targets.contains(jid.as_str()));
+        }
+        Ok(())
+    }
+
     // --- LID-PN Mapping ---
 
     async fn get_lid_mapping(&self, lid: &str) -> Result<Option<LidPnMappingEntry>> {

wacore/src/store/persistence.rs

@@ -210,4 +210,13 @@ impl PersistenceManager {
     pub async fn clear_sender_key_devices(&self, group_jid: &str) -> Result<(), StoreError> {
         self.backend.clear_sender_key_devices(group_jid).await
     }
+
+    pub async fn delete_sender_key_device_rows(
+        &self,
+        device_jids: &[&str],
+    ) -> Result<(), StoreError> {
+        self.backend
+            .delete_sender_key_device_rows(device_jids)
+            .await
+    }
 }

wacore/src/store/traits.rs

@@ -230,6 +230,10 @@ pub trait ProtocolStore: Send + Sync {
     /// Clear all sender key device tracking for a group (on sender key rotation).
     async fn clear_sender_key_devices(&self, group_jid: &str) -> Result<()>;
 
+    /// Delete specific `sender_key_devices` rows by device JID across all groups.
+    /// Mirrors WA Web's per-group `senderKey.delete(deviceJid)` cleanup.
+    async fn delete_sender_key_device_rows(&self, device_jids: &[&str]) -> Result<()>;
+
     /// Clear all sender key device tracking across ALL groups.
     /// Called on identity change (raw_id mismatch) to force SKDM redistribution.
     async fn clear_all_sender_key_devices(&self) -> Result<()>;