perf(send): parallelize group encrypt fan-out + adjacent wins

src/client/device_registry.rs

@@ -190,6 +190,65 @@ impl Client {
         Ok(())
     }
 
+    /// Batched variant of [`update_device_list`]. Cache is populated
+    /// synchronously per record (cheap moka inserts); the backend write
+    /// 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.
+    pub(crate) async fn update_device_lists(
+        &self,
+        records: Vec<wacore::store::traits::DeviceListRecord>,
+    ) -> Result<()> {
+        use anyhow::Context;
+
+        if records.is_empty() {
+            return Ok(());
+        }
+
+        let mut prepared = Vec::with_capacity(records.len());
+        let mut to_delete: Vec<String> = Vec::new();
+
+        for mut record in records {
+            let original_user = record.user.clone();
+            let lookup = self.resolve_lookup_keys(&original_user).await;
+            let canonical_key = lookup.canonical_key().to_string();
+            record.user.clone_from(&canonical_key);
+
+            let record_for_cache = record.clone();
+            self.device_registry_cache
+                .insert(canonical_key.clone(), record_for_cache)
+                .await;
+
+            if canonical_key != original_user {
+                to_delete.push(original_user);
+            }
+            prepared.push(record);
+        }
+
+        let backend = self.persistence_manager.backend();
+        backend
+            .update_device_lists(prepared)
+            .await
+            .context("Failed to update device lists in backend")?;
+
+        // Canonical-flip cleanup is rare and per-row; keep the original
+        // pattern (invalidate cache + best-effort delete + re-invalidate)
+        // 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;
+            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;
+        }
+
+        Ok(())
+    }
+
     /// Invalidate cached device data for a specific user.
     ///
     /// Removes all device registry cache entries (all LID/PN aliases) so the

src/features/signal.rs

@@ -273,6 +273,7 @@ impl<'a> Signal<'a> {
 
         let mut stores = adapter.as_signal_stores();
         let result = wacore::send::encrypt_for_devices(
+            &*self.client.runtime,
             &mut stores,
             self.client,
             &device_jids,

src/send.rs

@@ -414,6 +414,7 @@ impl Client {
         };
 
         let prepared = match wacore::send::prepare_group_stanza(
+            &*self.runtime,
             &mut stores,
             self,
             &mut group_info,
@@ -455,6 +456,7 @@ impl Client {
                     let mut stores_retry = store_adapter_retry.as_signal_stores();
 
                     wacore::send::prepare_group_stanza(
+                        &*self.runtime,
                         &mut stores_retry,
                         self,
                         &mut group_info,
@@ -951,11 +953,10 @@ impl Client {
             )
             .await?
         } else if to.is_group() {
-            // Group messages: No client-level lock needed.
-            // Each participant device is encrypted separately with its own per-device lock
-            // inside prepare_group_stanza, so we don't need to serialize entire group sends.
-
-            // Preparation work (no lock needed)
+            // Group messages: no client-level lock needed. The encrypt fan-out
+            // inside prepare_group_stanza touches a different Signal session
+            // per recipient device, so concurrent group sends to the same
+            // chat don't race on shared state.
             let mut group_info = self.groups().query_info(&to).await?;
 
             let mut device_snapshot = self.persistence_manager.get_device_snapshot().await;
@@ -1044,6 +1045,7 @@ impl Client {
             };
 
             match wacore::send::prepare_group_stanza(
+                &*self.runtime,
                 &mut stores,
                 self,
                 &mut group_info,
@@ -1088,6 +1090,7 @@ impl Client {
                         let mut stores_retry = store_adapter_retry.as_signal_stores();
 
                         let retry_prepared = wacore::send::prepare_group_stanza(
+                            &*self.runtime,
                             &mut stores_retry,
                             self,
                             &mut group_info,
@@ -1242,6 +1245,7 @@ impl Client {
             let mut stores = store_adapter.as_signal_stores();
 
             let prepared = wacore::send::prepare_dm_stanza(
+                &*self.runtime,
                 &mut stores,
                 self,
                 own_jid,

src/store/signal_adapter.rs

@@ -156,14 +156,16 @@ impl IdentityKeyStore for IdentityAdapter {
 
     async fn is_trusted_identity(
         &self,
-        address: &ProtocolAddress,
-        identity: &IdentityKey,
-        direction: Direction,
+        _address: &ProtocolAddress,
+        _identity: &IdentityKey,
+        _direction: Direction,
     ) -> Result<bool, SignalProtocolError> {
-        let device = self.0.device.read().await;
-        IdentityKeyStore::is_trusted_identity(&*device, address, identity, direction)
-            .await
-            .map_err(signal_err("is_trusted_identity"))
+        // WAWebProtocolStoreUnifiedApi.isTrustedIdentity always returns true;
+        // identity changes surface via save_identity. Avoid acquiring the
+        // device RwLock just to delegate to a stub — the read is acquired N
+        // times per group send (once per recipient device) and adds
+        // contention pressure under any future parallel encrypt path.
+        Ok(true)
     }
 
     async fn get_identity(

src/usync.rs

@@ -57,6 +57,8 @@ impl Client {
             }
 
             let mut fetched_devices = Vec::with_capacity(response.device_lists.len());
+            let mut device_records: Vec<wacore::store::traits::DeviceListRecord> =
+                Vec::with_capacity(response.device_lists.len());
 
             for user_list in &response.device_lists {
                 // Update device registry (single source of truth for device lists).
@@ -141,19 +143,20 @@ impl Client {
                     fetched_devices.push(jid);
                 }
 
-                let device_list = wacore::store::traits::DeviceListRecord {
+                device_records.push(wacore::store::traits::DeviceListRecord {
                     user: user_list.user.user.to_string(),
                     devices,
                     timestamp: wacore::time::now_secs(),
                     phash: user_list.phash.clone(),
                     raw_id,
-                };
-                if let Err(e) = self.update_device_list(device_list).await {
-                    warn!(
-                        "Failed to update device registry for {}: {}",
-                        user_list.user.user, e
-                    );
-                }
+                });
+            }
+
+            // One batched backend write for the whole usync response — for
+            // large groups this collapses N spawn_blocking SQLite hops into
+            // a single transaction, which dominated the per-send wall-clock.
+            if let Err(e) = self.update_device_lists(device_records).await {
+                warn!("Failed to update device registry batch: {e}");
             }
 
             all_devices.extend(fetched_devices);

storages/sqlite-storage/src/sqlite_store.rs

@@ -2156,6 +2156,73 @@ impl ProtocolStore for SqliteStore {
         Ok(())
     }
 
+    async fn update_device_lists(&self, records: Vec<DeviceListRecord>) -> Result<()> {
+        if records.is_empty() {
+            return Ok(());
+        }
+        let device_id = self.device_id;
+        let now = wacore::time::now_secs() as i32;
+
+        // Pre-serialize devices_json once (outside the retry loop and outside
+        // spawn_blocking) so retries are zero-allocation. Each row carries its
+        // own json+raw_id alongside the record.
+        struct PreparedRow {
+            user: String,
+            devices_json: String,
+            timestamp: i32,
+            phash: Option<String>,
+            raw_id: Option<i32>,
+        }
+
+        let prepared: Vec<PreparedRow> = records
+            .into_iter()
+            .map(|r| {
+                let devices_json = serde_json::to_string(&r.devices)
+                    .map_err(|e| StoreError::Serialization(Box::new(e)))?;
+                Ok(PreparedRow {
+                    user: r.user,
+                    devices_json,
+                    timestamp: r.timestamp as i32,
+                    phash: r.phash,
+                    raw_id: r.raw_id.map(|v| v as i32),
+                })
+            })
+            .collect::<Result<Vec<_>>>()?;
+        let prepared = std::sync::Arc::new(prepared);
+
+        self.with_retry("update_device_lists", move || {
+            let prepared = std::sync::Arc::clone(&prepared);
+            Box::new(move |conn: &mut SqliteConnection| {
+                conn.transaction::<_, DieselError, _>(|conn| {
+                    for row in prepared.iter() {
+                        diesel::insert_into(device_registry::table)
+                            .values((
+                                device_registry::user_id.eq(&row.user),
+                                device_registry::devices_json.eq(&row.devices_json),
+                                device_registry::timestamp.eq(row.timestamp),
+                                device_registry::phash.eq(&row.phash),
+                                device_registry::device_id.eq(device_id),
+                                device_registry::updated_at.eq(now),
+                                device_registry::raw_id.eq(row.raw_id),
+                            ))
+                            .on_conflict((device_registry::user_id, device_registry::device_id))
+                            .do_update()
+                            .set((
+                                device_registry::devices_json.eq(&row.devices_json),
+                                device_registry::timestamp.eq(row.timestamp),
+                                device_registry::phash.eq(&row.phash),
+                                device_registry::updated_at.eq(now),
+                                device_registry::raw_id.eq(row.raw_id),
+                            ))
+                            .execute(conn)?;
+                    }
+                    Ok(())
+                })
+            })
+        })
+        .await
+    }
+
     async fn get_devices(&self, user: &str) -> Result<Option<DeviceListRecord>> {
         let pool = self.pool.clone();
         let device_id = self.device_id;

wacore/Cargo.toml

@@ -55,7 +55,7 @@ waproto = { workspace = true }
 
 [dev-dependencies]
 aes-gcm = { workspace = true }
-futures = { workspace = true, features = ["executor"] }
+futures = { workspace = true, features = ["executor", "thread-pool"] }
 iai-callgrind = { workspace = true }
 tokio = { workspace = true, features = ["macros", "rt"] }