perf(send): memoize the group phash on the device-list memo entry

src/client/device_registry.rs

@@ -24,7 +24,7 @@ pub(crate) struct GroupDevicesMemo {
     /// counterparts, resolved device users): the scoped-invalidation check
     /// tests the topology log's touched users against this set.
     pub(crate) members: Arc<std::collections::HashSet<wacore_binary::CompactString>>,
-    pub(crate) devices: Arc<Vec<Jid>>,
+    pub(crate) devices: Arc<wacore::send::ResolvedGroupDevices>,
 }
 
 /// Result of resolving a user identifier to lookup keys.
@@ -92,16 +92,16 @@ impl Client {
         group: &Jid,
         group_info: &Arc<wacore::client::context::GroupInfo>,
         own_sending_jid: &Jid,
-    ) -> Result<Arc<Vec<Jid>>, anyhow::Error> {
+    ) -> Result<Arc<wacore::send::ResolvedGroupDevices>, anyhow::Error> {
         // Store-backed registry or mapping caches can be written by OTHER
         // processes (e.g. shared Redis across pods), which this process's
         // topology tracker cannot observe; the memo's freshness contract
         // doesn't hold there, so it is disabled and every send resolves.
         if !self.group_devices_memo_enabled {
-            return Ok(Arc::new(
+            return Ok(Arc::new(wacore::send::ResolvedGroupDevices::new(
                 self.resolve_group_devices_uncached(group_info, own_sending_jid)
                     .await?,
-            ));
+            )));
         }
         // Load the generation BEFORE resolving (do NOT move this after
         // get_user_devices): a write racing the resolve bumps it afterwards,
@@ -169,7 +169,7 @@ impl Client {
             members.insert(device.user.clone());
         }
 
-        let devices = Arc::new(devices);
+        let devices = Arc::new(wacore::send::ResolvedGroupDevices::new(devices));
         self.group_devices_memo
             .insert(
                 group.clone(),
@@ -1119,7 +1119,7 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve should succeed");
-        assert_eq!(first.len(), 3, "0+5 for A, 0 for B");
+        assert_eq!(first.devices().len(), 3, "0+5 for A, 0 for B");
 
         // Raw cache write WITHOUT a topology bump: the memo must keep serving
         // the snapshot (this is what proves the repeat call was a hit and not
@@ -1129,8 +1129,8 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve should succeed");
-        assert_eq!(
-            stale, first,
+        assert!(
+            std::sync::Arc::ptr_eq(&stale, &first),
             "same Arc + same generation must be a memo hit"
         );
 
@@ -1141,7 +1141,11 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve should succeed");
-        assert_eq!(fresh.len(), 2, "post-bump resolve must see the raw change");
+        assert_eq!(
+            fresh.devices().len(),
+            2,
+            "post-bump resolve must see the raw change"
+        );
 
         // A refreshed GroupInfo (new Arc, identical content) must recompute
         // even with an unchanged generation.
@@ -1159,7 +1163,7 @@ mod tests {
             .await
             .expect("resolve should succeed");
         assert_eq!(
-            after_refresh.len(),
+            after_refresh.devices().len(),
             3,
             "a new GroupInfo Arc must invalidate the memo by identity"
         );
@@ -1183,7 +1187,7 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve");
-        assert_eq!(first.len(), 2);
+        assert_eq!(first.devices().len(), 2);
 
         // Raw change (not recorded) + changes touching only a NON-member:
         // the memo must re-stamp and keep serving the snapshot.
@@ -1194,8 +1198,8 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve");
-        assert_eq!(
-            stale, first,
+        assert!(
+            std::sync::Arc::ptr_eq(&stale, &first),
             "non-member changes must re-stamp, not recompute"
         );
 
@@ -1205,7 +1209,7 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve");
-        assert_eq!(fresh.len(), 1, "member change must recompute");
+        assert_eq!(fresh.devices().len(), 1, "member change must recompute");
 
         // Global events (mapping cache clear, warm-up) poison the fast path.
         setup_device_record(&client, user_a, &[0, 5, 9]).await;
@@ -1214,7 +1218,11 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve");
-        assert_eq!(after_global.len(), 3, "global event must recompute");
+        assert_eq!(
+            after_global.devices().len(),
+            3,
+            "global event must recompute"
+        );
 
         // Log overflow past the memo's stamp: cannot prove cleanliness,
         // must recompute.
@@ -1226,7 +1234,11 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve");
-        assert_eq!(after_overflow.len(), 1, "log overflow must recompute");
+        assert_eq!(
+            after_overflow.devices().len(),
+            1,
+            "log overflow must recompute"
+        );
     }
 
     /// A mapping add for a member (logged under BOTH its LID and PN keys)
@@ -1246,7 +1258,7 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve");
-        assert_eq!(first.len(), 1);
+        assert_eq!(first.devices().len(), 1);
 
         // Raw change, then learn a LID mapping for the member: the add logs
         // (lid, pn) and the memo's member set carries the PN, so it must
@@ -1265,7 +1277,7 @@ mod tests {
             .await
             .expect("resolve");
         assert_eq!(
-            fresh.len(),
+            fresh.devices().len(),
             2,
             "a member's mapping change must invalidate the memo"
         );
@@ -1294,7 +1306,11 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &own)
             .await
             .expect("resolve");
-        assert_eq!(first.len(), 3, "member device + own's two devices");
+        assert_eq!(
+            first.devices().len(),
+            3,
+            "member device + own's two devices"
+        );
 
         let second = client
             .resolve_group_devices_memoized(&group, &group_info, &own)
@@ -1346,7 +1362,7 @@ mod tests {
             .resolve_group_devices_memoized(&group, &group_info, &group_info.participants[0])
             .await
             .expect("resolve");
-        assert_eq!(first.len(), 1);
+        assert_eq!(first.devices().len(), 1);
 
         // The update arrives keyed by the LID: canonical == original == LID,
         // so without the alias rule only the LID would be recorded and the
@@ -1376,7 +1392,7 @@ mod tests {
             .await
             .expect("resolve");
         assert_eq!(
-            fresh.len(),
+            fresh.devices().len(),
             2,
             "a LID-keyed update for a member must invalidate the PN group's memo"
         );

src/send.rs

@@ -869,7 +869,7 @@ impl Client {
         group_jid: &str,
         group_info: &wacore::client::context::GroupInfo,
         own_sending_jid: &Jid,
-    ) -> Option<(std::sync::Arc<Vec<Jid>>, Vec<Jid>)> {
+    ) -> Option<(std::sync::Arc<wacore::send::ResolvedGroupDevices>, Vec<Jid>)> {
         let cached_map = self.skdm_device_map(group_jid).await;
 
         let is_lid_mode = group_info.addressing_mode == wacore::types::message::AddressingMode::Lid;
@@ -897,9 +897,14 @@ impl Client {
                 } else {
                     all_devices
                 };
-                let all_devices = std::sync::Arc::new(all_devices);
-                let needs_skdm =
-                    self.filter_skdm_targets(group_jid, &all_devices, &cached_map, own_sending_jid);
+                let all_devices =
+                    std::sync::Arc::new(wacore::send::ResolvedGroupDevices::new(all_devices));
+                let needs_skdm = self.filter_skdm_targets(
+                    group_jid,
+                    all_devices.devices(),
+                    &cached_map,
+                    own_sending_jid,
+                );
                 Some((all_devices, needs_skdm))
             }
             Err(e) => {
@@ -923,15 +928,19 @@ impl Client {
         group_jid: &str,
         group_info: &std::sync::Arc<wacore::client::context::GroupInfo>,
         own_sending_jid: &Jid,
-    ) -> Option<(std::sync::Arc<Vec<Jid>>, Vec<Jid>)> {
+    ) -> Option<(std::sync::Arc<wacore::send::ResolvedGroupDevices>, Vec<Jid>)> {
         let cached_map = self.skdm_device_map(group_jid).await;
         match self
             .resolve_group_devices_memoized(group, group_info, own_sending_jid)
             .await
         {
             Ok(all_devices) => {
-                let needs_skdm =
-                    self.filter_skdm_targets(group_jid, &all_devices, &cached_map, own_sending_jid);
+                let needs_skdm = self.filter_skdm_targets(
+                    group_jid,
+                    all_devices.devices(),
+                    &cached_map,
+                    own_sending_jid,
+                );
                 Some((all_devices, needs_skdm))
             }
             Err(e) => {
@@ -1452,7 +1461,7 @@ impl Client {
             // still missing the key. On the cold/`force_skdm` path both are
             // `None` and `prepare_group_stanza` resolves the set itself.
             let (all_devices_for_phash, skdm_target_devices): (
-                Option<std::sync::Arc<Vec<Jid>>>,
+                Option<std::sync::Arc<wacore::send::ResolvedGroupDevices>>,
                 Option<Vec<Jid>>,
             ) = if force_skdm {
                 (None, None)
@@ -2970,10 +2979,10 @@ mod tests {
         // Empty cache → every participant needs SKDM, and the full set equals
         // the target set on this cold path.
         assert_eq!(needs_skdm.len(), participants.len());
-        assert_eq!(all_devices.len(), participants.len());
+        assert_eq!(all_devices.devices().len(), participants.len());
         for user in &participant_users {
             assert!(needs_skdm.iter().any(|j| j.user == *user));
-            assert!(all_devices.iter().any(|j| j.user == *user));
+            assert!(all_devices.devices().iter().any(|j| j.user == *user));
         }
     }
 

wacore/benches/send_receive_benchmark.rs

@@ -594,6 +594,9 @@ struct GrpSendData {
     alice: User,
     group_jid: Jid,
     participants: Vec<Jid>,
+    /// Warm-send fixture: the resolved set with its phash memo pre-warmed in
+    /// setup, like the per-group device memo serves production repeat sends.
+    resolved_for_phash: Option<std::sync::Arc<wacore::send::ResolvedGroupDevices>>,
     force_skdm: bool,
     resolver: MockResolver,
     msg: wa::Message,
@@ -625,10 +628,18 @@ fn setup_group_send(n: usize) -> GrpSendData {
             .unwrap();
     });
 
+    let resolved = std::sync::Arc::new(wacore::send::ResolvedGroupDevices::new(
+        participants.clone(),
+    ));
+    // Warm steady state: production warms the memo on the first send after a
+    // topology change and serves every later send from it.
+    let _ = resolved.phash(&alice.jid);
+
     GrpSendData {
         alice,
         group_jid,
         participants,
+        resolved_for_phash: Some(resolved),
         force_skdm: false,
         resolver: MockResolver(devices),
         msg: text_msg(),
@@ -650,16 +661,19 @@ fn setup_group_send_256() -> GrpSendData {
 fn setup_group_skdm_10() -> GrpSendData {
     let mut d = setup_group_send(10);
     d.force_skdm = true;
+    d.resolved_for_phash = None;
     d
 }
 fn setup_group_skdm_50() -> GrpSendData {
     let mut d = setup_group_send(50);
     d.force_skdm = true;
+    d.resolved_for_phash = None;
     d
 }
 fn setup_group_skdm_256() -> GrpSendData {
     let mut d = setup_group_send(256);
     d.force_skdm = true;
+    d.resolved_for_phash = None;
     d
 }
 
@@ -775,8 +789,7 @@ fn run_group_send(d: &mut GrpSendData) {
     // only emits a phash if it gets the full device set. Mirror the real
     // warm-send caller by passing it; the cold/force_skdm path resolves the set
     // itself and keeps None.
-    let all_devices_for_phash =
-        (!d.force_skdm).then(|| std::sync::Arc::new(d.participants.clone()));
+    let all_devices_for_phash = d.resolved_for_phash.clone();
     let mut group_info = GroupInfo::new(std::mem::take(&mut d.participants), AddressingMode::Pn);
     let own_base = own_jid.to_non_ad();
     if !group_info

wacore/src/send.rs

@@ -21,7 +21,7 @@ use std::collections::HashSet;
 use std::future::Future;
 use wacore_binary::Node;
 use wacore_binary::builder::NodeBuilder;
-use wacore_binary::{Jid, JidExt as _};
+use wacore_binary::{CompactString, Jid, JidExt as _};
 use wacore_libsignal::crypto::aes_256_cbc_encrypt_into;
 use waproto::whatsapp as wa;
 
@@ -68,6 +68,7 @@ mod dm;
 mod encrypt;
 mod group;
 mod peer;
+mod resolved_devices;
 mod status;
 
 pub use classify::*;
@@ -78,6 +79,7 @@ pub use dm::*;
 pub use encrypt::*;
 pub use group::*;
 pub use peer::*;
+pub use resolved_devices::ResolvedGroupDevices;
 pub use status::*;
 
 #[cfg(test)]

wacore/src/send/group.rs

@@ -139,7 +139,7 @@ pub async fn prepare_group_stanza<
     // sends so the phash covers every device + self even when no SKDM is sent;
     // `None` on the cold `force_skdm` path (the set is resolved here) and for
     // status broadcasts (which keep the prior phash behavior).
-    all_devices_for_phash: Option<std::sync::Arc<Vec<Jid>>>,
+    all_devices_for_phash: Option<std::sync::Arc<super::ResolvedGroupDevices>>,
     edit: Option<crate::types::message::EditAttribute>,
     extra_stanza_nodes: &[Node],
 ) -> Result<PreparedGroupStanza> {
@@ -169,7 +169,7 @@ pub async fn prepare_group_stanza<
 
     let mut message_children: Vec<Node> = Vec::new();
     let mut includes_prekey_message = false;
-    let mut phash_for_stanza: Option<String> = None;
+    let mut phash_for_stanza: Option<CompactString> = None;
     let mut skdm_encrypted_devices: Vec<Jid> = Vec::new();
 
     // Determine if we need to distribute SKDM and to which devices.
@@ -302,17 +302,16 @@ pub async fn prepare_group_stanza<
     // broadcasts keep the prior behavior (phash over the distribution list only,
     // when distributing); WA Web's status path does not augment with self.
     if to_jid.is_group() {
-        // Warm/partial sends pass the complete set in `all_devices_for_phash`;
-        // the cold `force_skdm` path leaves it None and `distribution_list`
-        // already holds the full resolved set.
-        if let Some(src) = all_devices_for_phash
-            .as_deref()
-            .map(Vec::as_slice)
-            .or(distribution_list.as_deref())
-        {
+        // Warm/partial sends pass the complete set in `all_devices_for_phash`,
+        // whose phash memo serves repeat sends with an inline copy; the cold
+        // `force_skdm` path leaves it None and `distribution_list` already
+        // holds the full resolved set.
+        if let Some(resolved) = all_devices_for_phash.as_deref() {
+            phash_for_stanza = resolved.phash(&own_sending_jid);
+        } else if let Some(src) = distribution_list.as_deref() {
             let phash_set = build_group_phash_set(src, &own_sending_jid);
             match MessageUtils::participant_list_hash(&phash_set) {
-                Ok(phash) => phash_for_stanza = Some(phash),
+                Ok(phash) => phash_for_stanza = Some(CompactString::new(&phash)),
                 Err(e) => {
                     log::warn!(
                         "Failed to compute group phash for {}: {:?}",
@@ -324,7 +323,7 @@ pub async fn prepare_group_stanza<
         }
     } else if let Some(ref distribution_list) = distribution_list {
         match MessageUtils::participant_list_hash(distribution_list) {
-            Ok(phash) => phash_for_stanza = Some(phash),
+            Ok(phash) => phash_for_stanza = Some(CompactString::new(&phash)),
             Err(e) => log::warn!("Failed to compute phash for {}: {:?}", to_jid.observe(), e),
         }
     }