feat(send): add targeted retransmission controls

src/cache.rs

@@ -5,3 +5,16 @@
 //! `get_with`) that builds on every target, including wasm32.
 
 pub use crate::portable_cache::PortableCache as Cache;
+
+/// Selects whether an operation may use an existing snapshot or must refresh it
+/// from its authoritative source before returning.
+#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
+#[non_exhaustive]
+pub enum Freshness {
+    /// Return a cached snapshot when available and consult the source on a miss.
+    #[default]
+    CachePreferred,
+    /// Consult the source and publish the resulting snapshot without clearing the
+    /// previous one first.
+    Refresh,
+}

src/client/app_state.rs

@@ -1030,12 +1030,11 @@ impl Client {
                     self.send_message_impl(
                         peer,
                         msg,
-                        Some(self.generate_message_id()),
-                        true,
-                        false,
-                        None,
-                        Vec::new(),
-                        None,
+                        crate::send::SendPipelineOptions {
+                            request_id: Some(self.generate_message_id()),
+                            peer: true,
+                            ..Default::default()
+                        },
                     )
                     .await
                 }

src/client/device_registry.rs

@@ -132,8 +132,12 @@ impl Client {
         // doesn't hold there, so it is disabled and every send resolves.
         if !self.group_devices_memo_enabled {
             return Ok(Arc::new(wacore::send::ResolvedGroupDevices::new(
-                self.resolve_group_devices_uncached(group_info, own_sending_jid)
-                    .await?,
+                self.resolve_group_devices_uncached(
+                    group_info,
+                    own_sending_jid,
+                    crate::cache::Freshness::CachePreferred,
+                )
+                .await?,
             )));
         }
         // Load the generation BEFORE resolving (do NOT move this after
@@ -175,7 +179,11 @@ impl Client {
         }
 
         let devices = self
-            .resolve_group_devices_uncached(group_info, own_sending_jid)
+            .resolve_group_devices_uncached(
+                group_info,
+                own_sending_jid,
+                crate::cache::Freshness::CachePreferred,
+            )
             .await?;
 
         // Member identifiers in both namespaces, so the scoped-invalidation
@@ -221,10 +229,11 @@ impl Client {
     /// (participants + LID normalization, appending self when the server
     /// snapshot omitted it — mirroring `ensure_self_in_group`, so keying the
     /// memo off the pre-ensure Arc stays equivalent) and resolve it.
-    async fn resolve_group_devices_uncached(
+    pub(crate) async fn resolve_group_devices_uncached(
         &self,
-        group_info: &Arc<wacore::client::context::GroupInfo>,
+        group_info: &wacore::client::context::GroupInfo,
         own_sending_jid: &Jid,
+        freshness: crate::cache::Freshness,
     ) -> Result<Vec<Jid>, anyhow::Error> {
         let is_lid_mode = group_info.addressing_mode == wacore::types::message::AddressingMode::Lid;
         let mut jids_to_resolve: Vec<Jid> = group_info
@@ -256,7 +265,12 @@ impl Client {
             jids_to_resolve.push(own);
         }
 
-        let mut devices = self.get_user_devices(&jids_to_resolve).await?;
+        let mut devices = match freshness {
+            crate::cache::Freshness::CachePreferred => {
+                self.get_user_devices_owned(jids_to_resolve).await?
+            }
+            crate::cache::Freshness::Refresh => self.refresh_user_devices(jids_to_resolve).await?,
+        };
         if is_lid_mode {
             // WA Web expects LID addressing in SKDM <to> nodes for LID groups.
             devices = devices
@@ -623,8 +637,7 @@ impl Client {
                 } else {
                     // Filter stale devices by valid_indexes. A raw_id reset already
                     // removed every companion while preserving primary metadata.
-                    record.devices =
-                        wacore::adv::filter_devices_by_key_index(&record.devices, &decoded);
+                    wacore::adv::retain_devices_by_key_index(&mut record.devices, &decoded);
                 }
                 record.raw_id = Some(decoded.raw_id);
 

src/client/messaging.rs

@@ -108,12 +108,10 @@ impl Client {
         self.send_message_impl(
             to,
             &edit_container_message,
-            None,
-            false,
-            false,
-            Some(crate::types::message::EditAttribute::MessageEdit),
-            vec![],
-            None,
+            crate::send::SendPipelineOptions {
+                edit: Some(crate::types::message::EditAttribute::MessageEdit),
+                ..Default::default()
+            },
         )
         .await
         .map_err(crate::send::SendError::from_anyhow)?;
@@ -196,12 +194,10 @@ impl Client {
         self.send_message_impl(
             to,
             &envelope,
-            None,
-            false,
-            false,
-            Some(crate::types::message::EditAttribute::MessageEdit),
-            vec![],
-            None,
+            crate::send::SendPipelineOptions {
+                edit: Some(crate::types::message::EditAttribute::MessageEdit),
+                ..Default::default()
+            },
         )
         .await
         .map_err(SendError::from_anyhow)?;

src/client/sender_keys.rs

@@ -30,20 +30,20 @@ impl Client {
             .and_then(|s| s.pn.as_ref())
             .map(|j| j.user.as_str());
 
-        let kept: Vec<&Jid> = device_jids
+        let keep = |jid: &&Jid| {
+            !exclude_own_devices
+                || !(own_lid_user.is_some_and(|user| user == jid.user)
+                    || own_pn_user.is_some_and(|user| user == jid.user))
+        };
+        let device_ids: Vec<String> = device_jids
             .iter()
-            .filter(|jid| {
-                !exclude_own_devices
-                    || !(own_lid_user.is_some_and(|u| u == jid.user)
-                        || own_pn_user.is_some_and(|u| u == jid.user))
-            })
+            .filter(keep)
+            .map(ToString::to_string)
             .collect();
-
-        if kept.is_empty() {
+        if device_ids.is_empty() {
             return Ok(());
         }
 
-        let device_ids: Vec<String> = kept.iter().map(|jid| jid.to_string()).collect();
         let entries: Vec<(&str, bool)> = device_ids.iter().map(|s| (s.as_str(), has_key)).collect();
         self.persistence_manager
             .set_sender_key_status(group_jid, &entries)
@@ -61,9 +61,8 @@ impl Client {
             // the skdm_warm_memo compares, so a warm send re-runs its target
             // filter and re-sends the now-cold device's SKDM — no separate memo
             // invalidation, hence no cross-cache ordering window.
-            let jids: Vec<Jid> = kept.into_iter().cloned().collect();
             self.sender_key_device_cache
-                .mark_forgotten(group_jid, &jids)
+                .mark_forgotten(group_jid, device_jids.iter().filter(keep))
                 .await;
         }
         Ok(())

src/features/groups.rs

@@ -299,7 +299,24 @@ impl<'a> Groups<'a> {
     }
 
     pub async fn query_info(&self, jid: &Jid) -> Result<Arc<GroupInfo>, GroupError> {
-        if let Some(cached) = self.client.get_group_cache().await.get(jid).await {
+        self.query_info_with_freshness(jid, crate::cache::Freshness::CachePreferred)
+            .await
+    }
+
+    /// Query group metadata using the requested cache freshness policy.
+    ///
+    /// A refresh leaves the current snapshot readable while the network request
+    /// is in flight, then atomically replaces it after a successful response.
+    pub async fn query_info_with_freshness(
+        &self,
+        jid: &Jid,
+        freshness: crate::cache::Freshness,
+    ) -> Result<Arc<GroupInfo>, GroupError> {
+        let cache = self.client.get_group_cache().await;
+        let cached = cache.get(jid).await;
+        if freshness == crate::cache::Freshness::CachePreferred
+            && let Some(cached) = cached
+        {
             return Ok(cached);
         }
 
@@ -308,11 +325,18 @@ impl<'a> Groups<'a> {
         // group, letting us reuse the persisted metadata instead of re-parsing it.
         let jid_str = jid.to_string();
         let backend = self.client.persistence_manager.backend();
-        let persisted: Option<GroupInfo> = match backend.get_group_metadata(&jid_str).await {
-            Ok(Some(blob)) => serde_json::from_slice(&blob).ok(),
-            _ => None,
+        // A refresh with a warm L1 can use that immutable snapshot directly as
+        // the not-modified fallback, avoiding a DB read and deserialize.
+        let persisted: Option<GroupInfo> = if cached.is_some() {
+            None
+        } else {
+            match backend.get_group_metadata(&jid_str).await {
+                Ok(Some(blob)) => serde_json::from_slice(&blob).ok(),
+                _ => None,
+            }
         };
-        let phash = persisted.as_ref().and_then(|info| {
+        let phash_source = cached.as_deref().or(persisted.as_ref());
+        let phash = phash_source.and_then(|info| {
             wacore::messages::MessageUtils::participant_list_hash(&info.participants).ok()
         });
 
@@ -322,16 +346,15 @@ impl<'a> Groups<'a> {
             .await?
         {
             GroupInfoOutcome::NotModified => {
-                let info = Arc::new(persisted.ok_or_else(|| {
-                    GroupError::InvalidRequest(
-                        "server returned not-modified group but nothing was cached".into(),
-                    )
-                })?);
-                self.client
-                    .get_group_cache()
-                    .await
-                    .insert(jid.clone(), info.clone())
-                    .await;
+                let info = match cached {
+                    Some(cached) => cached,
+                    None => Arc::new(persisted.ok_or_else(|| {
+                        GroupError::InvalidRequest(
+                            "server returned not-modified group but nothing was cached".into(),
+                        )
+                    })?),
+                };
+                cache.insert(jid.clone(), info.clone()).await;
                 return Ok(info);
             }
             GroupInfoOutcome::Full(group) => *group,
@@ -395,11 +418,7 @@ impl<'a> Groups<'a> {
         }
 
         let info = Arc::new(info);
-        self.client
-            .get_group_cache()
-            .await
-            .insert(jid.clone(), info.clone())
-            .await;
+        cache.insert(jid.clone(), info.clone()).await;
 
         Ok(info)
     }
@@ -1147,12 +1166,11 @@ impl<'a> Groups<'a> {
             .send_message_impl(
                 group_jid.clone(),
                 &msg,
-                Some(message_id.clone()),
-                false,
-                false,
-                None,
-                meta.into_iter().collect(),
-                None,
+                crate::send::SendPipelineOptions {
+                    request_id: Some(message_id.clone()),
+                    extra_stanza_nodes: meta.into_iter().collect(),
+                    ..Default::default()
+                },
             )
             .await?;
         Ok(message_id)
@@ -1536,6 +1554,33 @@ mod tests {
         assert_eq!(a.participants.len(), 2);
     }
 
+    #[tokio::test]
+    async fn refresh_keeps_the_previous_group_snapshot_on_source_failure() {
+        let client = crate::test_utils::create_test_client().await;
+        let group: Jid = "120363000000000099@g.us".parse().unwrap();
+        let previous = Arc::new(GroupInfo::new(
+            vec!["12025550101@s.whatsapp.net".parse().unwrap()],
+            AddressingMode::Pn,
+        ));
+        let cache = client.get_group_cache().await;
+        cache.insert(group.clone(), Arc::clone(&previous)).await;
+
+        let result = client
+            .groups()
+            .query_info_with_freshness(&group, crate::cache::Freshness::Refresh)
+            .await;
+        assert!(
+            result.is_err(),
+            "the offline fixture proves refresh consulted the source"
+        );
+
+        let preserved = cache
+            .get(&group)
+            .await
+            .expect("refresh failure must not clear the current snapshot");
+        assert!(Arc::ptr_eq(&previous, &preserved));
+    }
+
     #[tokio::test]
     async fn linked_removal_preserves_unrelated_group_cache_entries() {
         use wacore::protocol::ProtocolNode;
@@ -1544,7 +1589,7 @@ mod tests {
         let client = crate::test_utils::create_test_client().await;
         let parent: Jid = "120363000000000001@g.us".parse().unwrap();
         let unrelated: Jid = "120363000000000002@g.us".parse().unwrap();
-        let removed: Jid = "15550000001@s.whatsapp.net".parse().unwrap();
+        let removed: Jid = "12025550103@s.whatsapp.net".parse().unwrap();
         let cache = client.get_group_cache().await;
         for jid in [&parent, &unrelated] {
             cache

src/features/mod.rs

@@ -80,8 +80,8 @@ pub use status::{Status, StatusPrivacySetting, StatusSendOptions};
 pub use signal::{Signal, SignalError, SignalSessionInfo, SignalSessionMigration};
 pub(crate) use stanza::required_stanza_attr;
 pub use stanza::{
-    NackReason, RetryReason, RetryRequestError, RetryRequestOptions, RetryRequestOutcome,
-    StanzaRejection, StanzaResponseError,
+    MessageRetransmission, NackReason, RetryReason, RetryRequestError, RetryRequestOptions,
+    RetryRequestOutcome, StanzaRejection, StanzaResponseError,
 };
 pub use wacore::message_processing::EncType;