fix: codebase audit with bug fixes, race condition mitigations, and perf improvements

src/cache_config.rs

@@ -178,11 +178,11 @@ pub struct CacheConfig {
     pub sender_key_devices_cache: CacheEntryConfig,
 
     // --- Coordination caches (capacity-only, no TTL) ---
-    /// Per-device Signal session lock capacity. Default: 2000.
+    /// Per-device Signal session lock capacity. Default: 10000.
     pub session_locks_capacity: u64,
-    /// Per-chat message processing queue capacity. Default: 2000.
+    /// Per-chat message processing queue capacity. Default: 5000.
     pub message_queues_capacity: u64,
-    /// Per-chat message enqueue lock capacity. Default: 2000.
+    /// Per-chat message enqueue lock capacity. Default: 5000.
     pub message_enqueue_locks_capacity: u64,
 
     // --- Sent message DB cleanup ---
@@ -253,9 +253,12 @@ impl Default for CacheConfig {
             message_retry_counts: CacheEntryConfig::new(five_min, 1_000),
             pdo_pending_requests: CacheEntryConfig::new(Some(Duration::from_secs(30)), 500),
             sender_key_devices_cache: CacheEntryConfig::new(one_hour, 500),
-            session_locks_capacity: 2_000,
-            message_queues_capacity: 2_000,
-            message_enqueue_locks_capacity: 2_000,
+            // Coordination caches hold live mutexes/senders; capacity eviction
+            // while a reference is held creates a second lock for the same key,
+            // breaking serialization. Size generously to avoid eviction pressure.
+            session_locks_capacity: 10_000,
+            message_queues_capacity: 5_000,
+            message_enqueue_locks_capacity: 5_000,
             sent_message_ttl_secs: 300,
             cache_stores: CacheStores::default(),
         }

src/client.rs

@@ -2586,8 +2586,15 @@ impl Client {
                     // Evict stale entries to prevent unbounded growth over long sessions
                     guard.retain(|_, t| t.elapsed() < std::time::Duration::from_secs(24 * 3600));
                     drop(guard);
-                    if !to_request.is_empty() {
-                        self.request_app_state_keys(&to_request).await;
+                    if !to_request.is_empty()
+                        && let Err(e) = self.request_app_state_keys(&to_request).await
+                    {
+                        warn!("Failed to send app state key request: {e}");
+                        // Remove stamps so these keys can be retried next sync
+                        let mut guard = self.app_state_key_requests.lock().await;
+                        for key_id in &to_request {
+                            guard.remove(&hex::encode(key_id));
+                        }
                     }
                 }
 
@@ -2793,8 +2800,15 @@ impl Client {
                 // Evict stale entries to prevent unbounded growth over long sessions
                 guard.retain(|_, t| t.elapsed() < std::time::Duration::from_secs(24 * 3600));
                 drop(guard);
-                if !to_request.is_empty() {
-                    self.request_app_state_keys(&to_request).await;
+                if !to_request.is_empty()
+                    && let Err(e) = self.request_app_state_keys(&to_request).await
+                {
+                    warn!("Failed to send app state key request: {e}");
+                    // Remove stamps so these keys can be retried next sync
+                    let mut guard = self.app_state_key_requests.lock().await;
+                    for key_id in &to_request {
+                        guard.remove(&hex::encode(key_id));
+                    }
                 }
             }
 
@@ -2816,14 +2830,18 @@ impl Client {
         Ok(())
     }
 
-    async fn request_app_state_keys(&self, raw_key_ids: &[Vec<u8>]) {
+    async fn request_app_state_keys(&self, raw_key_ids: &[Vec<u8>]) -> Result<(), anyhow::Error> {
         if raw_key_ids.is_empty() {
-            return;
+            return Ok(());
         }
         let device_snapshot = self.persistence_manager.get_device_snapshot().await;
         let own_jid = match device_snapshot.pn.clone() {
             Some(j) => j,
-            None => return,
+            None => {
+                return Err(anyhow::anyhow!(
+                    "no own JID available for app-state key request"
+                ));
+            }
         };
         let key_ids: Vec<wa::message::AppStateSyncKeyId> = raw_key_ids
             .iter()
@@ -2839,20 +2857,17 @@ impl Client {
             })),
             ..Default::default()
         };
-        if let Err(e) = self
-            .send_message_impl(
-                own_jid,
-                &msg,
-                Some(self.generate_message_id().await),
-                true,
-                false,
-                None,
-                vec![],
-            )
-            .await
-        {
-            warn!("Failed to send app state key request: {e}");
-        }
+        self.send_message_impl(
+            own_jid,
+            &msg,
+            Some(self.generate_message_id().await),
+            true,
+            false,
+            None,
+            vec![],
+        )
+        .await?;
+        Ok(())
     }
 
     /// Send an app state patch to the server for a given collection.

src/client/sender_keys.rs

@@ -83,7 +83,9 @@ impl Client {
                 let mid = key.id.clone();
                 self.runtime
                     .spawn(Box::pin(async move {
-                        let _ = backend.take_sent_message(&cs, &mid).await;
+                        if let Err(e) = backend.take_sent_message(&cs, &mid).await {
+                            log::warn!("Failed to clean up sent message {cs}:{mid}: {e}");
+                        }
                     }))
                     .detach();
                 return Some(msg);

src/features/community.rs

@@ -172,8 +172,8 @@ impl<'a> Community<'a> {
             .execute(LinkSubgroupsIq::new(community_jid, subgroup_jids))
             .await?;
 
-        let mut linked_jids = Vec::new();
-        let mut failed_groups = Vec::new();
+        let mut linked_jids = Vec::with_capacity(response.groups.len());
+        let mut failed_groups = Vec::with_capacity(response.groups.len());
 
         for group in response.groups {
             if let Some(error) = group.error {
@@ -205,8 +205,8 @@ impl<'a> Community<'a> {
             ))
             .await?;
 
-        let mut unlinked_jids = Vec::new();
-        let mut failed_groups = Vec::new();
+        let mut unlinked_jids = Vec::with_capacity(response.groups.len());
+        let mut failed_groups = Vec::with_capacity(response.groups.len());
 
         for group in response.groups {
             if let Some(error) = group.error {
@@ -292,11 +292,13 @@ impl<'a> Community<'a> {
             .ok_or_else(|| MexError::PayloadParsing("missing data field".into()))?;
 
         let group_query = &data["xwa2_group_query_by_id"];
-        let mut counts = Vec::new();
+        let edges_ref = group_query
+            .get("sub_groups")
+            .and_then(|s| s.get("edges"))
+            .and_then(|e| e.as_array());
+        let mut counts = Vec::with_capacity(edges_ref.map_or(0, |e| e.len()));
 
-        if let Some(sub_groups) = group_query.get("sub_groups")
-            && let Some(edges) = sub_groups.get("edges").and_then(|e| e.as_array())
-        {
+        if let Some(edges) = edges_ref {
             for edge in edges {
                 if let Some(node) = edge.get("node") {
                     let id_str = node["id"].as_str().unwrap_or_default();

src/handlers/message.rs

@@ -66,6 +66,9 @@ impl StanzaHandler for MessageHandler {
                 let (tx, rx) = async_channel::unbounded::<Arc<Node>>();
 
                 let client_for_worker = client.clone();
+                let spawn_generation = client
+                    .connection_generation
+                    .load(std::sync::atomic::Ordering::Acquire);
 
                 // Spawn a worker task that processes messages sequentially for this chat.
                 // The worker exits when all tx senders are dropped (cache TTI expiry drops
@@ -76,6 +79,15 @@ impl StanzaHandler for MessageHandler {
                     .runtime
                     .spawn(Box::pin(async move {
                         while let Ok(msg_node) = rx.recv().await {
+                            // Exit if the connection changed — prevents stale workers
+                            // from processing messages with outdated crypto state.
+                            if client_for_worker
+                                .connection_generation
+                                .load(std::sync::atomic::Ordering::Acquire)
+                                != spawn_generation
+                            {
+                                break;
+                            }
                             let start = wacore::time::now_millis() as u64;
                             let client = client_for_worker.clone();
                             Box::pin(client.handle_incoming_message(msg_node)).await;

src/handlers/notification.rs

@@ -148,6 +148,16 @@ async fn handle_notification_impl(client: &Arc<Client>, node: &Node) {
                             log::debug!(target: "Client/AppState", "Skipping server_sync: client is shutting down");
                             return;
                         }
+                        // Re-check generation after version filtering to avoid syncing
+                        // against a stale connection after the awaited work above.
+                        if client_clone
+                            .connection_generation
+                            .load(std::sync::atomic::Ordering::SeqCst)
+                            != generation
+                        {
+                            log::debug!(target: "Client/AppState", "server_sync task cancelled: connection generation changed during version check");
+                            return;
+                        }
                         if let Err(e) = client_clone.sync_collections_batched(to_sync).await
                             && !client_clone.is_shutting_down()
                         {

src/send.rs

@@ -483,20 +483,26 @@ impl Client {
             .ensure_status_participants(prepared.node, &group_info)
             .await?;
 
-        let our_phash = stanza
+        let ack = if let Some(phash) = stanza
             .attrs()
             .optional_string("phash")
-            .map(|s| s.into_owned());
-        let ack_rx = if our_phash.is_some() {
-            Some(self.register_ack_waiter(&request_id).await)
+            .map(|s| s.into_owned())
+        {
+            let rx = self.register_ack_waiter(&request_id).await;
+            Some((rx, phash))
         } else {
             None
         };
 
-        self.send_node(stanza).await?;
+        if let Err(e) = self.send_node(stanza).await {
+            if ack.is_some() {
+                self.response_waiters.lock().await.remove(&request_id);
+            }
+            return Err(e.into());
+        }
 
-        if let Some(rx) = ack_rx {
-            self.spawn_phash_validation(rx, our_phash.unwrap(), to.clone(), false);
+        if let Some((rx, phash)) = ack {
+            self.spawn_phash_validation(rx, phash, to.clone(), false, request_id.clone());
         }
 
         self.update_sender_key_devices(&to_str, &prepared.skdm_devices)
@@ -635,6 +641,7 @@ impl Client {
         our_phash: String,
         jid: Jid,
         invalidate_group_cache: bool,
+        message_id: String,
     ) {
         let Some(client) = self.self_weak.get().and_then(|w| w.upgrade()) else {
             return;
@@ -648,7 +655,11 @@ impl Client {
                 .await
                 {
                     Ok(Ok(node)) => node,
-                    _ => return,
+                    _ => {
+                        // Remove leaked waiter to prevent keepalive suppression
+                        client.response_waiters.lock().await.remove(&message_id);
+                        return;
+                    }
                 };
                 if let Some(server) = ack.attrs().optional_string("phash")
                     && *server != our_phash
@@ -1104,24 +1115,31 @@ impl Client {
             .await?
         };
 
-        let our_phash = stanza_to_send
+        let ack = if let Some(phash) = stanza_to_send
             .attrs()
             .optional_string("phash")
-            .map(|s| s.into_owned());
-        let ack_rx = if our_phash.is_some() {
-            let msg_id = stanza_to_send.attrs().optional_string("id");
-            Some(
-                self.register_ack_waiter(msg_id.as_deref().unwrap_or_default())
-                    .await,
-            )
+            .map(|s| s.into_owned())
+        {
+            let msg_id = stanza_to_send
+                .attrs()
+                .optional_string("id")
+                .map(|s| s.into_owned())
+                .unwrap_or_default();
+            let rx = self.register_ack_waiter(&msg_id).await;
+            Some((rx, phash, msg_id))
         } else {
             None
         };
 
-        self.send_node(stanza_to_send).await?;
+        if let Err(e) = self.send_node(stanza_to_send).await {
+            if let Some((_, _, ref msg_id)) = ack {
+                self.response_waiters.lock().await.remove(msg_id);
+            }
+            return Err(e.into());
+        }
 
-        if let Some(rx) = ack_rx {
-            self.spawn_phash_validation(rx, our_phash.unwrap(), tc_issue_target.clone(), true);
+        if let Some((rx, phash, msg_id)) = ack {
+            self.spawn_phash_validation(rx, phash, tc_issue_target.clone(), true, msg_id);
         }
 
         if let Some(update) = skdm_update {

src/upload.rs

@@ -164,9 +164,17 @@ where
                             });
                         }
                         UploadExistsResult::Resume { byte_offset } => {
-                            log::info!("Resuming upload from byte {byte_offset}/{total}");
-                            upload_data = &enc.data_to_upload[byte_offset as usize..];
-                            file_offset = Some(byte_offset);
+                            let offset = byte_offset as usize;
+                            if offset > enc.data_to_upload.len() {
+                                log::warn!(
+                                    "Server resume offset {offset} exceeds data length {}; uploading from start",
+                                    enc.data_to_upload.len()
+                                );
+                            } else {
+                                log::info!("Resuming upload from byte {byte_offset}/{total}");
+                                upload_data = &enc.data_to_upload[offset..];
+                                file_offset = Some(byte_offset);
+                            }
                         }
                         UploadExistsResult::NotFound => {}
                     }

src/usync.rs

@@ -56,7 +56,7 @@ impl Client {
                 );
             }
 
-            let mut fetched_devices = Vec::new();
+            let mut fetched_devices = Vec::with_capacity(response.device_lists.len());
 
             for user_list in &response.device_lists {
                 // Update device registry (single source of truth for device lists).