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
@@ -394,8 +408,17 @@ impl Client {
         )
     )]
     pub(crate) async fn update_device_list(
+        &self,
+        record: wacore::store::traits::DeviceListRecord,
+    ) -> Result<()> {
+        let guard = self.device_topology.lock_registry().await;
+        self.update_device_list_guarded(record, &guard).await
+    }
+
+    pub(crate) async fn update_device_list_guarded(
         &self,
         mut record: wacore::store::traits::DeviceListRecord,
+        guard: &crate::client::device_topology::DeviceRegistryMutationGuard<'_>,
     ) -> Result<()> {
         use anyhow::Context;
 
@@ -413,6 +436,7 @@ impl Client {
         // (whose member set only knows the PN side) would re-stamp stale.
         self.device_registry_cache
             .insert(
+                guard,
                 canonical_key.clone(),
                 Arc::new(record_for_cache),
                 lookup
@@ -433,14 +457,18 @@ impl Client {
             // gets cleared. Run the second invalidate unconditionally: even
             // if delete fails, the cache may have been repopulated with data
             // that no longer reflects our intent.
-            self.device_registry_cache.invalidate(&original_user).await;
+            self.device_registry_cache
+                .invalidate(guard, &original_user)
+                .await;
             if let Err(e) = backend.delete_devices(&original_user).await {
                 warn!(
                     "Failed to delete stale device row under {} after canonical flip: {e}",
                     original_user
                 );
             }
-            self.device_registry_cache.invalidate(&original_user).await;
+            self.device_registry_cache
+                .invalidate(guard, &original_user)
+                .await;
             debug!(
                 "Device registry: stored under LID {} (resolved from {})",
                 canonical_key, original_user
@@ -455,10 +483,20 @@ impl Client {
     /// collapses into a single transaction. Used by usync after fetching
     /// device lists for many users at once, where the per-row commit
     /// dominated wall-clock time on large groups.
+    #[cfg(test)]
     #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.session.update_device_lists", level = "debug", skip_all, fields(count = records.len()), err(Debug)))]
     pub(crate) async fn update_device_lists(
         &self,
         records: Vec<wacore::store::traits::DeviceListRecord>,
+    ) -> Result<()> {
+        let guard = self.device_topology.lock_registry().await;
+        self.update_device_lists_guarded(records, &guard).await
+    }
+
+    pub(crate) async fn update_device_lists_guarded(
+        &self,
+        records: Vec<wacore::store::traits::DeviceListRecord>,
+        guard: &crate::client::device_topology::DeviceRegistryMutationGuard<'_>,
     ) -> Result<()> {
         use anyhow::Context;
 
@@ -479,6 +517,7 @@ impl Client {
             // Same alias rule as update_device_list: record every lookup key.
             self.device_registry_cache
                 .insert(
+                    guard,
                     canonical_key.clone(),
                     Arc::new(record_for_cache),
                     lookup
@@ -504,14 +543,18 @@ impl Client {
         // rather than batching deletes. On error we log and continue so a
         // single bad row doesn't drop the rest of the batch.
         for original_user in to_delete {
-            self.device_registry_cache.invalidate(&original_user).await;
+            self.device_registry_cache
+                .invalidate(guard, &original_user)
+                .await;
             if let Err(e) = backend.delete_devices(&original_user).await {
                 warn!(
                     "Failed to delete stale device row under {} after canonical flip: {e}",
                     original_user
                 );
             }
-            self.device_registry_cache.invalidate(&original_user).await;
+            self.device_registry_cache
+                .invalidate(guard, &original_user)
+                .await;
         }
 
         Ok(())
@@ -553,10 +596,19 @@ impl Client {
         )
     )]
     pub(crate) async fn invalidate_device_cache(&self, user: &str) {
+        let guard = self.device_topology.lock_registry().await;
+        self.invalidate_device_cache_guarded(user, &guard).await;
+    }
+
+    pub(crate) async fn invalidate_device_cache_guarded(
+        &self,
+        user: &str,
+        guard: &crate::client::device_topology::DeviceRegistryMutationGuard<'_>,
+    ) {
         let lookup = self.resolve_lookup_keys(user).await;
 
         for key in lookup.all_keys() {
-            self.device_registry_cache.invalidate(key).await;
+            self.device_registry_cache.invalidate(guard, key).await;
             // Also delete from DB so get_devices_from_registry doesn't
             // fall back to stale persisted data — forces a network re-fetch
             if let Err(e) = self.persistence_manager.backend().delete_devices(key).await {
@@ -566,7 +618,7 @@ impl Client {
             // the doomed DB row can promote() it back between the first
             // invalidate and the delete commit (same guard as the canonical
             // flip path in update_device_list).
-            self.device_registry_cache.invalidate(key).await;
+            self.device_registry_cache.invalidate(guard, key).await;
         }
 
         debug!("Invalidated device cache for user: {} ({:?})", user, lookup);
@@ -595,6 +647,7 @@ impl Client {
         device: &wacore::stanza::devices::DeviceElement,
         key_index_info: Option<&wacore::stanza::devices::KeyIndexInfo>,
     ) {
+        let guard = self.device_topology.lock_registry().await;
         let device_id = device.device_id();
         let is_hosted = wacore_binary::JidExt::is_hosted(&device.jid);
 
@@ -623,8 +676,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);
 
@@ -666,7 +718,7 @@ impl Client {
         // unknown device → `device_has_key()` returns `None` → falls into
         // `needs_skdm`. No global cache invalidation needed.
 
-        if let Err(e) = self.update_device_list(record).await {
+        if let Err(e) = self.update_device_list_guarded(record, &guard).await {
             warn!("patch_device_add: failed to persist: {e}");
         }
     }
@@ -763,6 +815,7 @@ impl Client {
         if device_id == 0 {
             return;
         }
+        let guard = self.device_topology.lock_registry().await;
         if let Some(mut record) = self.load_device_record(user).await {
             let before = record.devices.len();
             record.devices.retain(|d| d.device_id != device_id);
@@ -775,7 +828,7 @@ impl Client {
                         "patch_device_remove: device_id {device_id} > u16::MAX — skipping \
                          session/SKDM cleanup but still persisting registry removal"
                     );
-                    if let Err(e) = self.update_device_list(record).await {
+                    if let Err(e) = self.update_device_list_guarded(record, &guard).await {
                         warn!("patch_device_remove: failed to persist: {e}");
                     }
                     return;
@@ -800,7 +853,7 @@ impl Client {
                     );
                     return;
                 }
-                if let Err(e) = self.update_device_list(record).await {
+                if let Err(e) = self.update_device_list_guarded(record, &guard).await {
                     warn!("patch_device_remove: failed to persist: {e}");
                 }
             }
@@ -865,13 +918,14 @@ impl Client {
         user: &str,
         device: &wacore::stanza::devices::DeviceElement,
     ) {
+        let guard = self.device_topology.lock_registry().await;
         let device_id = device.device_id();
 
         if let Some(mut record) = self.load_device_record(user).await
             && let Some(d) = record.devices.iter_mut().find(|d| d.device_id == device_id)
         {
             d.key_index = device.key_index;
-            if let Err(e) = self.update_device_list(record).await {
+            if let Err(e) = self.update_device_list_guarded(record, &guard).await {
                 warn!("patch_device_update: failed to persist: {e}");
             }
         }
@@ -998,6 +1052,7 @@ impl Client {
         )
     )]
     pub(crate) async fn migrate_device_registry_on_lid_discovery(&self, pn: &str, lid: &str) {
+        let guard = self.device_topology.lock_registry().await;
         let backend = self.persistence_manager.backend();
 
         match backend.get_devices(pn).await {
@@ -1015,25 +1070,25 @@ impl Client {
                     // The backend row may have changed even on error, so the
                     // change is recorded before the early return; the success
                     // path records once via the fused cache insert below.
-                    self.device_topology.record([pn, lid]);
+                    self.device_topology.record_registry(&guard, [pn, lid]);
                     warn!("Failed to migrate device registry to LID: {}", e);
                     return;
                 }
 
                 self.device_registry_cache
-                    .insert(lid.to_string(), Arc::new(record), [lid, pn])
+                    .insert(&guard, lid.to_string(), Arc::new(record), [lid, pn])
                     .await;
 
                 // Drop the PN-keyed row in both cache and DB. Invalidate
                 // twice (before + after delete) so a concurrent reader can't
                 // resurrect the cache from the DB row between the two calls.
                 // Always run the second invalidate; even if delete fails, the
                 // cache may carry resurrected data that shouldn't stick.
-                self.device_registry_cache.invalidate(pn).await;
+                self.device_registry_cache.invalidate(&guard, pn).await;
                 if let Err(e) = backend.delete_devices(pn).await {
                     warn!("Failed to delete PN-keyed device row during LID migration: {e}");
                 }
-                self.device_registry_cache.invalidate(pn).await;
+                self.device_registry_cache.invalidate(&guard, pn).await;
             }
             Ok(None) => {}
             Err(e) => {