refactor: codebase quality audit — safety, tests, DRY, dep hygiene

Cargo.toml

@@ -70,6 +70,7 @@ sha1 = { version = "0.10.6", default-features = false }
 sha2 = { version = "0.10.6", default-features = false }
 thiserror = "2.0.17"
 tokio = { version = "1.48.0", default-features = false }
+uuid = { version = "1", default-features = false }
 
 # Internal workspace crates
 wacore = { path = "./wacore", default-features = false, version = "0.4.1" }
@@ -132,7 +133,7 @@ whatsapp-rust-tokio-transport = { path = "./transports/tokio-transport", version
 whatsapp-rust-ureq-http-client = { path = "./http_clients/ureq-client", version = "0.4.1", optional = true }
 
 [dev-dependencies]
-uuid = { version = "1.0", features = ["v4"] }
+uuid = { workspace = true, features = ["v4"] }
 
 [[example]]
 name = "benchmark"

src/appstate_sync.rs

@@ -2,12 +2,6 @@
 pub use wacore::appstate::Mutation;
 pub use wacore::appstate_sync::{AppStateProcessor, AppStateSyncDriver, AppStateSyncError};
 
-// The following re-exports are used by the test module
-#[allow(unused_imports)]
-use wacore::appstate::hash::HashState;
-#[allow(unused_imports)]
-use wacore::appstate::patch_decode::{PatchList, WAPatchName};
-
 #[cfg(test)]
 mod tests {
     use super::*;

src/client.rs

@@ -295,11 +295,11 @@ pub struct Client {
     /// and DB writes are deferred to flush() after each message is processed.
     pub(crate) signal_cache: Arc<crate::store::signal_cache::SignalStoreCache>,
 
-    /// Global semaphore that limits message processing concurrency.
-    /// During offline sync: permits=1 (sequential, like WA Web's allChatQueue)
-    /// After offline sync: permits=N (parallel per-chat processing)
-    /// Wrapped in std::sync::Mutex to allow replacing on reconnect.
+    /// Limits message processing concurrency (1 permit during offline sync, N after).
+    /// Wrapped in Mutex to allow replacing on reconnect.
     pub(crate) message_processing_semaphore: std::sync::Mutex<Arc<async_lock::Semaphore>>,
+    /// Bumped on every semaphore swap so stale Arc clones are rejected.
+    pub(crate) message_semaphore_generation: Arc<AtomicU64>,
 
     /// Per-device session locks for Signal protocol operations.
     /// Prevents race conditions when multiple messages from the same sender
@@ -557,6 +557,7 @@ impl Client {
             message_processing_semaphore: std::sync::Mutex::new(Arc::new(
                 async_lock::Semaphore::new(1),
             )),
+            message_semaphore_generation: Arc::new(AtomicU64::new(0)),
             // Coordination caches: capacity-only eviction, no TTL/TTI.
             // These hold live mutexes and channel senders; time-based eviction
             // while tasks hold references would silently break serialisation.
@@ -1012,10 +1013,17 @@ impl Client {
         self.retried_group_messages.invalidate_all();
         // Clear signal cache so stale state doesn't leak across connections
         self.signal_cache.clear().await;
-        // Reset message processing semaphore to 1 permit (sequential mode for next offline sync).
-        // Old workers holding the previous semaphore Arc will finish normally.
-        *self.message_processing_semaphore.lock().unwrap() =
-            Arc::new(async_lock::Semaphore::new(1));
+        // Reset semaphore to 1 permit for next offline sync.
+        // Generation bump invalidates stale Arc clones. Block scopes the !Send MutexGuard.
+        {
+            self.message_semaphore_generation
+                .fetch_add(1, Ordering::SeqCst);
+            let mut guard = match self.message_processing_semaphore.lock() {
+                Ok(g) => g,
+                Err(poisoned) => poisoned.into_inner(),
+            };
+            *guard = Arc::new(async_lock::Semaphore::new(1));
+        }
         // Reset dead-socket timestamps so stale values from the previous
         // connection don't trigger an immediate reconnect on the next one.
         self.last_data_received_ms.store(0, Ordering::Relaxed);
@@ -3783,11 +3791,10 @@ mod tests {
 
         let elapsed = start.elapsed();
         // Count available permits by trying to acquire non-blockingly
-        let semaphore = client
-            .message_processing_semaphore
-            .lock()
-            .expect("message_processing_semaphore poisoned")
-            .clone();
+        let semaphore = match client.message_processing_semaphore.lock() {
+            Ok(guard) => guard.clone(),
+            Err(poisoned) => poisoned.into_inner().clone(),
+        };
         let mut guards = Vec::new();
         while let Some(guard) = semaphore.try_acquire() {
             guards.push(guard);

src/client/sessions.rs

@@ -37,11 +37,15 @@ impl Client {
             // During offline sync, permits=1 serialized all message processing.
             // Replace with a new semaphore with 64 permits for concurrent processing.
             // Old workers holding the previous semaphore Arc will finish normally.
-            *self
-                .message_processing_semaphore
-                .lock()
-                .expect("message_processing_semaphore poisoned") =
-                std::sync::Arc::new(async_lock::Semaphore::new(64));
+            {
+                self.message_semaphore_generation
+                    .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                let mut guard = match self.message_processing_semaphore.lock() {
+                    Ok(g) => g,
+                    Err(poisoned) => poisoned.into_inner(),
+                };
+                *guard = std::sync::Arc::new(async_lock::Semaphore::new(64));
+            }
 
             self.offline_sync_notifier.notify(usize::MAX);
 

src/features/chat_actions.rs

@@ -19,7 +19,6 @@ use crate::client::Client;
 use anyhow::Result;
 use chrono::DateTime;
 use log::debug;
-use std::sync::Arc;
 use wacore::appstate::patch_decode::WAPatchName;
 use wacore::types::events::{
     ArchiveUpdate, ContactUpdate, Event, MarkChatAsReadUpdate, MuteUpdate, PinUpdate, StarUpdate,
@@ -208,13 +207,13 @@ pub(crate) fn dispatch_chat_mutation(
 
 /// Feature handle for chat management actions.
 ///
-/// Access via `client.chat_actions()` (requires `Arc<Client>`).
+/// Access via `client.chat_actions()`.
 pub struct ChatActions<'a> {
-    client: &'a Arc<Client>,
+    client: &'a Client,
 }
 
 impl<'a> ChatActions<'a> {
-    pub(crate) fn new(client: &'a Arc<Client>) -> Self {
+    pub(crate) fn new(client: &'a Client) -> Self {
         Self { client }
     }
 
@@ -450,9 +449,7 @@ impl<'a> ChatActions<'a> {
 
 impl Client {
     /// Access chat management actions (archive, pin, mute, star).
-    ///
-    /// Requires `Arc<Client>` because app state mutations need key access.
-    pub fn chat_actions(self: &Arc<Self>) -> ChatActions<'_> {
+    pub fn chat_actions(&self) -> ChatActions<'_> {
         ChatActions::new(self)
     }
 }

src/features/media_reupload.rs

@@ -9,7 +9,6 @@
 use crate::client::{Client, ClientError, NodeFilter};
 use anyhow::Result;
 use log::debug;
-use std::sync::Arc;
 use std::time::Duration;
 pub use wacore::media_retry::MediaRetryResult;
 use wacore::media_retry::{
@@ -34,11 +33,11 @@ pub struct MediaReuploadRequest<'a> {
 }
 
 pub struct MediaReupload<'a> {
-    client: &'a Arc<Client>,
+    client: &'a Client,
 }
 
 impl<'a> MediaReupload<'a> {
-    pub(crate) fn new(client: &'a Arc<Client>) -> Self {
+    pub(crate) fn new(client: &'a Client) -> Self {
         Self { client }
     }
 
@@ -115,7 +114,7 @@ impl<'a> MediaReupload<'a> {
 
 impl Client {
     /// Access media reupload operations.
-    pub fn media_reupload(self: &Arc<Self>) -> MediaReupload<'_> {
+    pub fn media_reupload(&self) -> MediaReupload<'_> {
         MediaReupload::new(self)
     }
 }

src/features/profile.rs

@@ -6,7 +6,6 @@ use crate::client::Client;
 use crate::store::commands::DeviceCommand;
 use anyhow::Result;
 use log::{debug, warn};
-use std::sync::Arc;
 use wacore::iq::contacts::SetProfilePictureSpec;
 use wacore::iq::profile::SetStatusTextSpec;
 use wacore_binary::builder::NodeBuilder;
@@ -15,11 +14,11 @@ pub use wacore::iq::contacts::SetProfilePictureResponse;
 
 /// Feature handle for profile operations.
 pub struct Profile<'a> {
-    client: &'a Arc<Client>,
+    client: &'a Client,
 }
 
 impl<'a> Profile<'a> {
-    pub(crate) fn new(client: &'a Arc<Client>) -> Self {
+    pub(crate) fn new(client: &'a Client) -> Self {
         Self { client }
     }
 
@@ -163,8 +162,8 @@ impl<'a> Profile<'a> {
 }
 
 impl Client {
-    /// Access profile operations (requires Arc<Client>).
-    pub fn profile(self: &Arc<Self>) -> Profile<'_> {
+    /// Access profile operations.
+    pub fn profile(&self) -> Profile<'_> {
         Profile::new(self)
     }
 }

src/message.rs

@@ -493,11 +493,33 @@ impl Client {
             );
         }
 
-        // Acquire global processing permit. During offline sync (1 permit), this serializes
-        // ALL message processing globally, matching WA Web's allChatQueue pattern.
-        // After offline sync completes, permits are increased for parallel processing.
-        let semaphore = self.message_processing_semaphore.lock().unwrap().clone();
+        // Acquire global processing permit (1 during offline sync, N after).
+        // Generation check rejects stale Arc clones from a previous connection.
+        let generation = self
+            .message_semaphore_generation
+            .load(std::sync::atomic::Ordering::SeqCst);
+        let semaphore = match self.message_processing_semaphore.lock() {
+            Ok(guard) => guard.clone(),
+            Err(poisoned) => poisoned.into_inner().clone(),
+        };
+        if generation
+            != self
+                .message_semaphore_generation
+                .load(std::sync::atomic::Ordering::SeqCst)
+        {
+            log::debug!("Stale semaphore generation, skipping message batch");
+            return;
+        }
         let _global_permit = semaphore.acquire_arc().await;
+        // Post-acquire recheck: generation could have changed during the .await
+        if generation
+            != self
+                .message_semaphore_generation
+                .load(std::sync::atomic::Ordering::SeqCst)
+        {
+            log::debug!("Semaphore generation changed during acquire, dropping stale permit");
+            return;
+        }
 
         log::debug!(
             "Starting PASS 1: Processing {} session establishment messages (pkmsg/msg)",

src/pair_code.rs

@@ -317,11 +317,8 @@ pub(crate) async fn handle_pair_code_notification(client: &Arc<Client>, node: &N
     // Get device keys
     let device_snapshot = client.persistence_manager.get_device_snapshot().await;
 
-    // Prepare encrypted key bundle
-    // TODO: Store `new_adv_secret` via DeviceCommand::SetAdvSecretKey to enable HMAC
-    // verification in pair-success. Currently the HMAC check in do_pair_crypto is
-    // commented out, so pairing works without it. See wacore/src/pair.rs:147-153.
-    let (wrapped_bundle, _new_adv_secret) = match PairCodeUtils::prepare_key_bundle(
+    // Prepare encrypted key bundle (includes rotated adv_secret_key)
+    let (wrapped_bundle, new_adv_secret) = match PairCodeUtils::prepare_key_bundle(
         &ephemeral_keypair,
         &primary_ephemeral_pub,
         &primary_identity_pub,
@@ -334,6 +331,14 @@ pub(crate) async fn handle_pair_code_notification(client: &Arc<Client>, node: &N
         }
     };
 
+    // Persist rotated adv_secret_key so HMAC verification works in pair-success.
+    client
+        .persistence_manager
+        .process_command(crate::store::commands::DeviceCommand::SetAdvSecretKey(
+            new_adv_secret,
+        ))
+        .await;
+
     // Build and send stage 2 IQ
     let req_id = client.generate_request_id();
     let identity_pub: [u8; 32] = device_snapshot

tests/e2e/Cargo.toml

@@ -14,7 +14,7 @@ dhat-heap = ["dep:dhat"]
 anyhow = { workspace = true }
 dhat = { version = "0.3", optional = true }
 tokio = { workspace = true, features = ["macros", "rt-multi-thread", "sync", "time"] }
-uuid = { version = "1.0", features = ["v4"] }
+uuid = { workspace = true, features = ["v4"] }
 wacore = { path = "../../wacore" }
 whatsapp-rust = { path = "../..", features = ["danger-skip-tls-verify", "debug-diagnostics"] }
 whatsapp-rust-sqlite-storage = { path = "../../storages/sqlite-storage" }

wacore/libsignal/Cargo.toml

@@ -22,7 +22,6 @@ ghash = "0.6.0"
 hex = { workspace = true }
 hkdf = { workspace = true }
 hmac = { workspace = true }
-itertools = { version = "0.14.0", default-features = false, features = ["use_alloc"] }
 log = { workspace = true }
 prost = { workspace = true }
 rand = { workspace = true }
@@ -31,12 +30,12 @@ sha1 = { workspace = true }
 sha2 = { workspace = true }
 subtle = "2.6.1"
 thiserror = { workspace = true }
-uuid = { version = "1.18.1", default-features = false }
+uuid = { workspace = true }
 waproto = { workspace = true }
 x25519-dalek = { version = "2.0.1", features = ["static_secrets"] }
 
 [dev-dependencies]
-futures = { version = "0.3", default-features = false, features = ["executor"] }
+futures = { workspace = true, features = ["executor"] }
 iai-callgrind = { workspace = true }
 
 [[bench]]