perf(signal): move backend I/O and user scans out of the global cache locks

src/client/adapters.rs

@@ -5,28 +5,25 @@ use anyhow::Context as _;
 
 impl Client {
     /// Build a [`SignalProtocolStoreAdapter`] from the current device state and signal cache.
-    pub(crate) async fn signal_adapter(
+    pub(crate) fn signal_adapter(
         &self,
     ) -> crate::store::signal_adapter::SignalProtocolStoreAdapter {
-        let device_store = self.persistence_manager.get_device_arc().await;
-        self.signal_adapter_from(device_store)
+        self.signal_adapter_from(self.persistence_manager.clone())
     }
 
     /// Build a standalone [`SenderKeyAdapter`] from the current device state and
     /// signal cache, avoiding the full five-store adapter on the SKDM path.
-    pub(crate) async fn sender_key_adapter(
-        &self,
-    ) -> crate::store::signal_adapter::SenderKeyAdapter {
+    pub(crate) fn sender_key_adapter(&self) -> crate::store::signal_adapter::SenderKeyAdapter {
         crate::store::signal_adapter::SenderKeyAdapter::new(
-            self.persistence_manager.get_device_arc().await,
+            self.persistence_manager.clone(),
             self.signal_cache.clone(),
         )
     }
 
-    /// Build a [`SignalProtocolStoreAdapter`] from a pre-fetched device arc.
+    /// Build a [`SignalProtocolStoreAdapter`] from a pre-fetched persistence handle.
     pub(crate) fn signal_adapter_from(
         &self,
-        device_store: Arc<RwLock<crate::store::Device>>,
+        device_store: Arc<PersistenceManager>,
     ) -> crate::store::signal_adapter::SignalProtocolStoreAdapter {
         crate::store::signal_adapter::SignalProtocolStoreAdapter::new(
             device_store,

src/client/device_registry.rs

@@ -3318,7 +3318,7 @@ mod tests {
             "rotation must wait for the in-flight advance"
         );
 
-        let mut sender_key_store = client.sender_key_adapter().await;
+        let mut sender_key_store = client.sender_key_adapter();
         group_encrypt(
             &mut sender_key_store,
             &name,

src/client/sessions.rs

@@ -485,7 +485,7 @@ impl Client {
             }
         }
 
-        let mut adapter = self.signal_adapter().await;
+        let mut adapter = self.signal_adapter();
         let mut rng = rand::make_rng::<StdRng>();
 
         let mut success_count = 0;
@@ -596,7 +596,7 @@ impl Client {
     /// pkmsg too, not as plain msg.
     #[cfg(feature = "voip-runtime")]
     pub(crate) async fn would_emit_pkmsg(&self, jid: &Jid) -> Result<bool, anyhow::Error> {
-        let device_store = self.persistence_manager.get_device_arc().await;
+        let device_store = self.persistence_manager.clone();
         let mut adapter = self.signal_adapter_from(device_store);
         let signal_addr = jid.to_protocol_address();
         wacore::send::pkmsg_would_be_emitted(&mut adapter.session_store, &signal_addr).await

src/features/signal.rs

@@ -182,7 +182,7 @@ impl<'a> Signal<'a> {
         let address = jid.to_protocol_address();
         let lock = self.client.session_lock_for(address.as_str()).await;
         let _guard = lock.lock().await;
-        let mut adapter = self.client.signal_adapter().await;
+        let mut adapter = self.client.signal_adapter();
         Ok(message_encrypt(
             plaintext,
             &address,
@@ -200,7 +200,7 @@ impl<'a> Signal<'a> {
         let address = jid.to_protocol_address();
         let lock = self.client.session_lock_for(address.as_str()).await;
         let _guard = lock.lock().await;
-        let mut adapter = self.client.signal_adapter().await;
+        let mut adapter = self.client.signal_adapter();
         let mut rng = rand::make_rng::<rand::rngs::StdRng>();
         let decrypted = message_decrypt(
             parsed,
@@ -242,7 +242,7 @@ impl<'a> Signal<'a> {
         bundle: &PreKeyBundle,
     ) -> Result<IdentityChange, SignalError> {
         let resolved = self.client.resolve_encryption_jid(jid).await;
-        let mut adapter = self.client.signal_adapter().await;
+        let mut adapter = self.client.signal_adapter();
         let mut rng = rand::make_rng::<rand::rngs::StdRng>();
         let identity_change = self
             .client
@@ -276,7 +276,7 @@ impl<'a> Signal<'a> {
         let distribution = decode_sender_key_distribution(distribution)?;
         let sender_address = sender_jid.to_non_ad().to_protocol_address();
         let sender_key_name = make_sender_key_name(group_jid, &sender_address);
-        let mut store = self.client.sender_key_adapter().await;
+        let mut store = self.client.sender_key_adapter();
         let chain_lock = store.sender_key_lock(&sender_key_name).await;
         let chain_guard = chain_lock.lock().await;
 
@@ -294,7 +294,7 @@ impl<'a> Signal<'a> {
     ) -> Result<Vec<u8>, SignalError> {
         let sender_address = sender_jid.to_non_ad().to_protocol_address();
         let sender_key_name = make_sender_key_name(group_jid, &sender_address);
-        let mut store = self.client.sender_key_adapter().await;
+        let mut store = self.client.sender_key_adapter();
         let chain_lock = store.sender_key_lock(&sender_key_name).await;
         let chain_guard = chain_lock.lock().await;
         let distribution = wacore::send::create_sender_key_distribution_message_for_group(
@@ -512,7 +512,7 @@ impl<'a> Signal<'a> {
             .await?
             .is_some();
 
-        let mut store = self.client.sender_key_adapter().await;
+        let mut store = self.client.sender_key_adapter();
         let mut rng = rand::make_rng::<rand::rngs::StdRng>();
 
         let pending_distribution = self
@@ -574,7 +574,7 @@ impl<'a> Signal<'a> {
         let sender_key_name =
             make_sender_key_name(group_jid, &sender_jid.to_non_ad().to_protocol_address());
 
-        let mut store = self.client.sender_key_adapter().await;
+        let mut store = self.client.sender_key_adapter();
         let chain_lock = store.sender_key_lock(&sender_key_name).await;
         let _chain_guard = chain_lock.lock().await;
 
@@ -652,7 +652,7 @@ impl<'a> Signal<'a> {
         let _session_guards = self.client.session_guards_for(&lock_jids).await;
 
         let plaintext = MessageUtils::encode_and_pad(message);
-        let mut adapter = self.client.signal_adapter().await;
+        let mut adapter = self.client.signal_adapter();
         let mediatype = wacore::send::media_type_from_message(message);
         let hide_decrypt_fail = wacore::send::should_hide_decrypt_fail(message);
 

src/message/receive.rs

@@ -638,7 +638,7 @@ impl Client {
         // Started after the lock so the histogram excludes lock/queue wait.
         let _t = wacore::telemetry::timer(wacore::telemetry::DECRYPT_DURATION);
 
-        let mut adapter = self.signal_adapter().await;
+        let mut adapter = self.signal_adapter();
         let mut rng = rand::make_rng::<rand::rngs::StdRng>();
         let mut outcome = SessionBatchOutcome::default();
         // Buffer plaintexts to handle after the ratchet lock drops (see the drain
@@ -1229,7 +1229,7 @@ impl Client {
         if payloads.is_empty() {
             return Ok(());
         }
-        let mut adapter = self.signal_adapter().await;
+        let mut adapter = self.signal_adapter();
 
         // Always use bare sender for sender key operations. Real WA delivers
         // skmsg with bare participant but pkmsg (SKDM) with device-qualified

src/message/tests.rs

@@ -778,7 +778,7 @@ async fn bobs_prekey_bundle_with_spk_id(client: &Arc<Client>, spk_id: u32) -> (P
     // Read/write prekeys through the same trait surface production uses
     // (see signal_adapter.rs). Avoids reaching past `PersistenceManager`
     // to mutate device storage directly.
-    let mut adapter = client.signal_adapter().await;
+    let mut adapter = client.signal_adapter();
     let spk_record = adapter
         .signed_pre_key_store
         .get_signed_pre_key(1.into())

src/retry.rs

@@ -839,7 +839,7 @@ impl Client {
         let signal_address = encryption_jid.to_protocol_address();
         let session_mutex = self.session_lock_for(signal_address.as_str()).await;
         let session_guard = session_mutex.lock().await;
-        let mut store_adapter = self.signal_adapter().await;
+        let mut store_adapter = self.signal_adapter();
         let device_snapshot = self.persistence_manager.get_device_snapshot();
         let edit = wacore::types::message::EditAttribute::infer_from_message(&message);
 
@@ -930,7 +930,7 @@ impl Client {
         let encoded = pre_encoded
             .filter(|_| can_reuse_encoding)
             .or(encoded_fallback.as_deref());
-        let device_store = self.persistence_manager.get_device_arc().await;
+        let device_store = self.persistence_manager.clone();
         let mut store_adapter = self.signal_adapter_from(device_store);
         let mut stores = store_adapter.as_signal_stores();
         let edit = wacore::types::message::EditAttribute::infer_from_message(&message);
@@ -1420,7 +1420,7 @@ impl Client {
             identity_key.into(),
         )?;
 
-        let mut adapter = self.signal_adapter().await;
+        let mut adapter = self.signal_adapter();
         let mut rng = rand::make_rng::<rand::rngs::StdRng>();
         self.install_prekey_bundle_cached(requester_jid, &bundle, &mut adapter, &mut rng)
             .await?;

src/send/mod.rs

@@ -1137,7 +1137,7 @@ impl Client {
         self.add_recent_message(&to, &request_id, &message, shared_content.clone())
             .await;
 
-        let device_store_arc = self.persistence_manager.get_device_arc().await;
+        let device_store_arc = self.persistence_manager.clone();
         let to_str = to.to_string();
         let distribution_guard = self.group_distribution_lock(&to).await;
 
@@ -1909,7 +1909,7 @@ impl Client {
             let session_mutex = self.session_lock_for(signal_addr.as_str()).await;
             let _session_guard = session_mutex.lock().await;
 
-            let mut store_adapter = self.signal_adapter().await;
+            let mut store_adapter = self.signal_adapter();
 
             let device_snapshot = self.persistence_manager.get_device_snapshot();
             wacore::send::prepare_peer_stanza(
@@ -1981,7 +1981,7 @@ impl Client {
                     .await;
             }
 
-            let device_store_arc = self.persistence_manager.get_device_arc().await;
+            let device_store_arc = self.persistence_manager.clone();
             let to_str = to.to_string();
 
             let (own_sending_jid, _) = match group_info.addressing_mode {
@@ -2427,7 +2427,7 @@ impl Client {
             let lock_jids = self.build_session_lock_keys(dm_devices.devices()).await;
             let _session_guards = self.session_guards_for(&lock_jids).await;
 
-            let mut store_adapter = self.signal_adapter().await;
+            let mut store_adapter = self.signal_adapter();
 
             let mut stores = store_adapter.as_signal_stores();
 
@@ -4907,7 +4907,7 @@ mod tests {
             .expect("prekey bundle task")
             .expect("prekey bundle");
 
-        let mut adapter = client.signal_adapter().await;
+        let mut adapter = client.signal_adapter();
         let mut rng = rand::make_rng::<rand::rngs::StdRng>();
         process_prekey_bundle(
             &peer.to_protocol_address(),
@@ -5124,7 +5124,7 @@ mod tests {
             .expect("prekey bundle task")
             .expect("prekey bundle");
         {
-            let mut adapter = client.signal_adapter().await;
+            let mut adapter = client.signal_adapter();
             let mut rng = rand::make_rng::<rand::rngs::StdRng>();
             process_prekey_bundle(
                 &lid_addr.to_protocol_address(),

src/store/persistence_manager.rs

@@ -74,7 +74,7 @@ impl PersistenceManager {
         })
     }
 
-    /// Handle for store adapters that need `&mut Device` trait access.
+    /// Handle for callers that need `&mut Device` trait access directly.
     /// For plain reads, prefer [`get_device_snapshot`](Self::get_device_snapshot).
     pub async fn get_device_arc(&self) -> Arc<RwLock<Device>> {
         self.device.clone()