perf(send): memoize the per-group device list behind a topology generation

src/client.rs

@@ -3,6 +3,7 @@ mod adapters;
 mod app_state;
 mod context_impl;
 mod device_registry;
+pub(crate) mod device_topology;
 mod iq_ops;
 mod lid_pn;
 mod lifecycle;
@@ -526,8 +527,24 @@ pub struct Client {
     /// LRU cache for device registry (matches WhatsApp Web's 5000 entry limit).
     /// Maps user ID to DeviceListRecord for fast device existence checks.
     /// Backed by persistent storage.
-    pub(crate) device_registry_cache:
-        TypedCache<String, Arc<wacore::store::traits::DeviceListRecord>>,
+    /// Device registry fused with its topology tracker: every write records
+    /// the change by construction, so the group-devices memo below can never
+    /// be left stale by a forgotten bump.
+    pub(crate) device_registry_cache: crate::client::device_topology::DeviceRegistryCache,
+    /// Shared topology tracker (generation + changed-users log). LidPnCache
+    /// records mapping changes into it; the memo validates against it.
+    pub(crate) device_topology: Arc<crate::client::device_topology::DeviceTopology>,
+    /// Whether the group-devices memo may be used: false when the registry
+    /// or LID-PN caches are store-backed (a shared external store can be
+    /// written by other processes, which the in-process topology tracker
+    /// cannot observe).
+    pub(crate) group_devices_memo_enabled: bool,
+    /// Per-group memo of the fully resolved (LID-converted) device list,
+    /// validated by GroupInfo identity + the device topology. Serves the
+    /// per-send full-set resolution in `resolve_skdm_targets` so a warm
+    /// repeat send skips the per-member cache fan-out.
+    pub(crate) group_devices_memo:
+        Cache<Jid, Arc<crate::client::device_registry::GroupDevicesMemo>>,
 
     /// Router for dispatching stanzas to their appropriate handlers
     pub(crate) stanza_router: crate::handlers::router::StanzaRouter,

src/client/device_topology.rs

@@ -0,0 +1,178 @@
+//! Device-topology change tracking for the per-group device-list memo.
+//!
+//! "Topology" here means anything that can change a device-list answer:
+//! registry record writes/invalidations and LID-PN mapping changes. Instead of
+//! trusting every write path to remember a manual generation bump, the bump
+//! lives INSIDE the write chokepoints ([`DeviceRegistryCache`] and
+//! `LidPnCache::add`), so a writer cannot forget it by construction.
+//!
+//! Each change also logs WHICH canonical users it touched (both namespaces),
+//! so a memo whose generation went stale can prove "none of the changed users
+//! are in my group" and re-stamp itself instead of recomputing. Every doubtful
+//! case (log overflow, global events) degrades to a recompute, never to
+//! serving stale data.
+
+use std::collections::VecDeque;
+use std::sync::Arc;
+use std::sync::atomic::Ordering;
+
+use portable_atomic::AtomicU64;
+use wacore_binary::CompactString;
+
+/// Bounded log capacity. Sized so a burst (e.g. a usync response for a large
+/// group) still fits; overflow just disables the scoped-revalidation fast
+/// path until affected memos recompute once.
+const TOPOLOGY_LOG_CAPACITY: usize = 256;
+
+struct TopologyLog {
+    /// (generation that the change produced, canonical user touched).
+    entries: VecDeque<(u64, CompactString)>,
+    /// Highest generation evicted from `entries` (0 = nothing evicted).
+    /// A memo older than this cannot be proven clean and must recompute.
+    floor: u64,
+}
+
+/// Shared tracker: a monotonic generation plus the bounded changed-users log.
+pub(crate) struct DeviceTopology {
+    generation: AtomicU64,
+    log: std::sync::Mutex<TopologyLog>,
+}
+
+impl DeviceTopology {
+    pub(crate) fn new() -> Arc<Self> {
+        Arc::new(Self {
+            generation: AtomicU64::new(0),
+            log: std::sync::Mutex::new(TopologyLog {
+                entries: VecDeque::with_capacity(TOPOLOGY_LOG_CAPACITY),
+                floor: 0,
+            }),
+        })
+    }
+
+    pub(crate) fn current(&self) -> u64 {
+        self.generation.load(Ordering::Acquire)
+    }
+
+    /// Record one topology change touching the given users (pass BOTH
+    /// namespaces of an identity when known: a mapping change alters which
+    /// canonical record either key resolves to).
+    pub(crate) fn record<'a>(&self, users: impl IntoIterator<Item = &'a str>) {
+        let mut log = self.log.lock().unwrap_or_else(|p| p.into_inner());
+        let generation = self.generation.load(Ordering::Acquire) + 1;
+        for user in users {
+            if log.entries.len() == TOPOLOGY_LOG_CAPACITY
+                && let Some((evicted_gen, _)) = log.entries.pop_front()
+            {
+                log.floor = evicted_gen;
+            }
+            log.entries
+                .push_back((generation, CompactString::from(user)));
+        }
+        // Publish the generation only after the log holds the users, so a
+        // reader that observes the new generation can always find (or rule
+        // out) the corresponding entries.
+        self.generation.store(generation, Ordering::Release);
+    }
+
+    /// Record a change whose blast radius is unknown (bulk warm-up, cache
+    /// clear): bumps and poisons the scoped fast path so every memo
+    /// recomputes once.
+    pub(crate) fn record_global(&self) {
+        let mut log = self.log.lock().unwrap_or_else(|p| p.into_inner());
+        let generation = self.generation.load(Ordering::Acquire) + 1;
+        log.entries.clear();
+        log.floor = generation;
+        self.generation.store(generation, Ordering::Release);
+    }
+
+    /// Whether every change after `since` only touched users for which
+    /// `is_member` returns false. `false` on any doubt (log overflow past
+    /// `since`), so callers recompute.
+    pub(crate) fn unchanged_for(&self, since: u64, is_member: impl Fn(&str) -> bool) -> bool {
+        let log = self.log.lock().unwrap_or_else(|p| p.into_inner());
+        if log.floor > since {
+            return false;
+        }
+        log.entries
+            .iter()
+            .filter(|(generation, _)| *generation > since)
+            .all(|(_, user)| !is_member(user))
+    }
+}
+
+/// The device registry cache plus its topology tracker, fused so every write
+/// records the change. Reads are pass-through; the only write entry points
+/// are [`insert`](Self::insert), [`invalidate`](Self::invalidate) and the
+/// non-recording [`promote`](Self::promote) (whose data is by definition what
+/// the DB fallback already answered).
+pub(crate) struct DeviceRegistryCache {
+    cache: crate::cache_store::TypedCache<String, Arc<wacore::store::traits::DeviceListRecord>>,
+    topology: Arc<DeviceTopology>,
+}
+
+impl DeviceRegistryCache {
+    pub(crate) fn new(
+        cache: crate::cache_store::TypedCache<String, Arc<wacore::store::traits::DeviceListRecord>>,
+        topology: Arc<DeviceTopology>,
+    ) -> Self {
+        Self { cache, topology }
+    }
+
+    pub(crate) async fn get(
+        &self,
+        key: &str,
+    ) -> Option<Arc<wacore::store::traits::DeviceListRecord>> {
+        self.cache.get(key).await
+    }
+
+    /// Write a record and log the touched users. `touched` carries the keys
+    /// whose answers change (canonical key, plus the original alias when the
+    /// canonical flipped).
+    pub(crate) async fn insert<'a>(
+        &self,
+        key: String,
+        record: Arc<wacore::store::traits::DeviceListRecord>,
+        touched: impl IntoIterator<Item = &'a str>,
+    ) {
+        self.cache.insert(key, record).await;
+        self.topology.record(touched);
+    }
+
+    pub(crate) async fn invalidate(&self, key: &str) {
+        self.cache.invalidate(key).await;
+        self.topology.record([key]);
+    }
+
+    /// Cache-fill from the DB row the fallback path would have returned: the
+    /// answer is unchanged, so no topology change is recorded.
+    pub(crate) async fn promote(
+        &self,
+        key: String,
+        record: Arc<wacore::store::traits::DeviceListRecord>,
+    ) {
+        self.cache.insert(key, record).await;
+    }
+
+    #[cfg(feature = "debug-diagnostics")]
+    pub(crate) fn entry_count(&self) -> u64 {
+        self.cache.entry_count()
+    }
+
+    /// Test-only passthrough for moka maintenance flushes.
+    #[cfg(test)]
+    pub(crate) async fn run_pending_tasks(&self) {
+        self.cache.run_pending_tasks().await;
+    }
+
+    /// Test-only raw write that bypasses topology recording, for fixture
+    /// seeding and for proving that memo hits really are hits (a raw change
+    /// must be served stale).
+    #[cfg(test)]
+    pub(crate) async fn raw_insert_for_tests(
+        &self,
+        key: String,
+        record: Arc<wacore::store::traits::DeviceListRecord>,
+    ) {
+        self.cache.insert(key, record).await;
+    }
+}

src/client/lifecycle.rs

@@ -2,6 +2,10 @@
 
 use super::*;
 
+/// Max groups with a cached resolved-device snapshot. LRU eviction covers
+/// accounts in more groups; an evicted entry just recomputes on next send.
+const GROUP_DEVICES_MEMO_CAPACITY: u64 = 64;
+
 impl Client {
     pub fn shutdown_signal(&self) -> wacore::runtime::ShutdownSignal {
         self.shutdown_notifier.subscribe()
@@ -107,6 +111,7 @@ impl Client {
 
         let (tx, rx) = async_channel::bounded(32);
 
+        let device_topology = crate::client::device_topology::DeviceTopology::new();
         let this = Self {
             runtime: runtime.clone(),
             core,
@@ -215,10 +220,19 @@ impl Client {
             custom_enc_handlers: std::sync::OnceLock::new(),
             chatstate_handlers: Arc::new(RwLock::new(Vec::new())),
             pdo_pending_requests: cache_config.pdo_pending_requests.build_with_ttl(),
-            device_registry_cache: cache_config.device_registry_cache.build_typed_ttl(
-                cache_config.cache_stores.device_registry_cache.clone(),
-                "device_registry",
+            device_registry_cache: crate::client::device_topology::DeviceRegistryCache::new(
+                cache_config.device_registry_cache.build_typed_ttl(
+                    cache_config.cache_stores.device_registry_cache.clone(),
+                    "device_registry",
+                ),
+                Arc::clone(&device_topology),
             ),
+            device_topology,
+            group_devices_memo_enabled: cache_config.cache_stores.device_registry_cache.is_none()
+                && cache_config.cache_stores.lid_pn_cache.is_none(),
+            group_devices_memo: Cache::builder()
+                .max_capacity(GROUP_DEVICES_MEMO_CAPACITY)
+                .build(),
             stanza_router: Self::create_stanza_router(),
             synchronous_ack: false,
             http_client,
@@ -232,6 +246,10 @@ impl Client {
         };
 
         let arc = Arc::new(this);
+        // Mapping changes alter which canonical record a device lookup
+        // resolves to, so LidPnCache records into the same topology tracker.
+        arc.lid_pn_cache
+            .attach_topology(Arc::clone(&arc.device_topology));
         let _ = arc.self_weak.set(Arc::downgrade(&arc));
 
         // Warm up the LID-PN cache from persistent storage

src/handlers/notification/mod.rs

@@ -815,7 +815,7 @@ mod tests {
         };
         client
             .device_registry_cache
-            .insert("5511999999999".into(), Arc::new(record))
+            .raw_insert_for_tests("5511999999999".into(), Arc::new(record))
             .await;
 
         // Seed a stored identity so the had-prior-identity gate runs the full reset
@@ -1118,7 +1118,7 @@ mod tests {
         // cleanup has something to do, but deliberately do NOT seed an identity.
         client
             .device_registry_cache
-            .insert(
+            .raw_insert_for_tests(
                 "5511666666666".into(),
                 Arc::new(wacore::store::traits::DeviceListRecord {
                     user: "5511666666666".into(),

src/lid_pn_cache.rs

@@ -46,6 +46,11 @@ pub struct LidPnCache {
     lid_to_entry: TypedCache<Arc<str>, Arc<LidPnEntry>>,
     /// Phone number -> Entry mapping (stores the most recent LID for that PN)
     pn_to_entry: TypedCache<Arc<str>, Arc<LidPnEntry>>,
+    /// Device-topology tracker (attached by Client construction): a mapping
+    /// change alters which canonical record either key resolves to, so adds
+    /// record both identifiers. Recording lives here, at the write
+    /// chokepoint, so callers cannot forget it.
+    topology: std::sync::OnceLock<Arc<crate::client::device_topology::DeviceTopology>>,
     /// PN -> the LID this process durably persisted for it. Lets the learn hot
     /// path skip a re-persist without swallowing the first live persist of a
     /// mapping an offline replay only warmed in memory. Keyed by the pair so a
@@ -85,15 +90,27 @@ impl LidPnCache {
                 // Always in-memory: tracks per-process persist state, never the
                 // mapping itself, so it must not go through the shared store.
                 persisted: TypedCache::from_moka(config.build_with_tti()),
+                topology: std::sync::OnceLock::new(),
             },
             None => Self {
                 lid_to_entry: TypedCache::from_moka(config.build_with_tti()),
                 pn_to_entry: TypedCache::from_moka(config.build_with_tti()),
                 persisted: TypedCache::from_moka(config.build_with_tti()),
+                topology: std::sync::OnceLock::new(),
             },
         }
     }
 
+    /// Attach the device-topology tracker. Mapping writes before the attach
+    /// (none in practice: Client construction attaches before warm-up) are
+    /// simply not scoped.
+    pub(crate) fn attach_topology(
+        &self,
+        topology: Arc<crate::client::device_topology::DeviceTopology>,
+    ) {
+        let _ = self.topology.set(topology);
+    }
+
     /// Returns approximate entry counts for the LID and PN maps.
     #[cfg(feature = "debug-diagnostics")]
     pub fn entry_counts(&self) -> (u64, u64) {
@@ -185,6 +202,9 @@ impl LidPnCache {
                 .insert(shared.phone_number.clone(), shared)
                 .await;
         }
+        if let Some(topology) = self.topology.get() {
+            topology.record([&*entry.lid, &*entry.phone_number]);
+        }
     }
 
     /// Whether this process has durably persisted exactly `phone -> lid`.
@@ -231,6 +251,9 @@ impl LidPnCache {
         self.lid_to_entry.clear().await;
         self.pn_to_entry.clear().await;
         self.persisted.clear().await;
+        if let Some(topology) = self.topology.get() {
+            topology.record_global();
+        }
     }
 
     /// Get the number of LID entries in the cache.