perf(client): hoist write-once state out of async mutexes

src/client.rs

@@ -973,7 +973,10 @@ pub struct Client {
     pub(crate) transport_events:
         Arc<Mutex<Option<async_channel::Receiver<crate::transport::TransportEvent>>>>,
     pub(crate) transport_factory: Arc<dyn crate::transport::TransportFactory>,
-    pub(crate) noise_socket: Arc<Mutex<Option<Arc<NoiseSocket>>>>,
+    /// Replaced per connection, so not a `OnceLock` — but every critical section
+    /// is a clone or a store, so a sync lock makes holding it across an `.await`
+    /// a compile error on the send path rather than a review question.
+    pub(crate) noise_socket: Arc<std::sync::Mutex<Option<Arc<NoiseSocket>>>>,
 
     /// Pending IQ/ack response waiters keyed by request id.
     ///
@@ -1027,7 +1030,9 @@ pub struct Client {
     pub(crate) lid_pn_cache: Arc<LidPnCache>,
     pub(crate) ab_props: Arc<wacore::store::ab_props::AbPropsCache>,
 
-    pub group_cache: Mutex<Option<Arc<GroupCache>>>,
+    /// Lazily built on the first group send and never replaced afterwards, so a
+    /// `OnceLock` keeps the read on that path down to an atomic load.
+    pub group_cache: std::sync::OnceLock<Arc<GroupCache>>,
 
     pub(crate) expected_disconnect: Arc<AtomicBool>,
     /// Set by `reconnect()` to suppress the "Message loop exited with an error" warning.
@@ -1093,7 +1098,9 @@ pub struct Client {
 
     pub(crate) needs_initial_full_sync: Arc<app_state::BootstrapGate>,
 
-    pub(crate) app_state_processor: Mutex<Option<Arc<AppStateProcessor>>>,
+    /// Built on first app-state use and never replaced: reconnect clears the
+    /// processor's key cache in place rather than swapping the processor.
+    pub(crate) app_state_processor: std::sync::OnceLock<Arc<AppStateProcessor>>,
     pub(crate) app_state_key_requests: Arc<Mutex<HashMap<Vec<u8>, wacore::time::Instant>>>,
     /// Tracks collections currently being synced to prevent duplicate sync tasks.
     /// Matches WA Web's in-flight tracking set in WAWebSyncdCollectionsStateMachine.
@@ -1159,7 +1166,7 @@ pub struct Client {
         )>,
     >,
     /// Contacts with active presence subscriptions that must be re-subscribed on reconnect.
-    pub(crate) presence_subscriptions: Arc<Mutex<HashSet<Jid>>>,
+    pub(crate) presence_subscriptions: Arc<std::sync::Mutex<HashSet<Jid>>>,
     /// Metrics for granular offline sync logging
     pub(crate) offline_sync_metrics: Arc<OfflineSyncMetrics>,
     /// Drives the WA Web pull-batch loop for offline backlog delivery.
@@ -1241,7 +1248,12 @@ pub struct Client {
 
     /// Chat state (typing indicator) handlers registered by external consumers.
     /// Each handler receives a `ChatStateEvent` describing the chat, optional participant and state.
-    pub(crate) chatstate_handlers: Arc<RwLock<Vec<ChatStateHandler>>>,
+    ///
+    /// Copy-on-write behind a sync lock, guarded by `chatstate_handler_count` so
+    /// the default (no handler registered) never takes the lock nor builds the
+    /// event that only a handler would read.
+    pub(crate) chatstate_handlers: Arc<std::sync::RwLock<Arc<[ChatStateHandler]>>>,
+    pub(crate) chatstate_handler_count: AtomicUsize,
 
     pub(crate) pdo_pending_requests: Cache<ChatMessageId, crate::pdo::PendingPdoRequest>,
 
@@ -1358,6 +1370,10 @@ pub struct Client {
     /// Keeps retry regressions deterministic without corrupting the test database.
     #[cfg(test)]
     pub(crate) app_state_key_share_prepare_test_failures: AtomicU32,
+    /// Counts `ChatStateEvent` constructions, so a test can prove the
+    /// no-handler fast path skips the build rather than just the invoke.
+    #[cfg(test)]
+    pub(crate) chatstate_events_built: AtomicU32,
 
     /// Holds the background saver's AbortHandle so the task lifetime follows
     /// `Arc<Client>` ref count instead of the Bot wrapper's. Set once by

src/client/accessors.rs

@@ -12,19 +12,15 @@ pub struct IdentityTags {
 }
 
 impl Client {
-    pub(crate) async fn get_group_cache(&self) -> Arc<GroupCache> {
-        let mut guard = self.group_cache.lock().await;
-        if let Some(cache) = guard.as_ref() {
-            return cache.clone();
-        }
-        debug!("Initializing Group Cache for the first time.");
-        let cache = Arc::new(
-            self.cache_config
-                .group_cache
-                .build_typed_ttl(self.cache_config.cache_stores.group_cache.clone(), "group"),
-        );
-        *guard = Some(cache.clone());
-        cache
+    pub(crate) fn get_group_cache(&self) -> &Arc<GroupCache> {
+        self.group_cache.get_or_init(|| {
+            debug!("Initializing Group Cache for the first time.");
+            Arc::new(
+                self.cache_config
+                    .group_cache
+                    .build_typed_ttl(self.cache_config.cache_stores.group_cache.clone(), "group"),
+            )
+        })
     }
 
     /// Subscribe an external event handler with an explicit event filter.
@@ -158,10 +154,9 @@ impl Client {
             .unwrap_or_else(|p| p.into_inner())
             .len();
 
-        // Only the Arc is taken under the mutex — the walk must not block
-        // get_group_cache(), which every group send goes through.
-        let group_cache_arc = self.group_cache.lock().await.clone();
-        let group_cache = match group_cache_arc {
+        // `get()`, not `get_group_cache()`: a report must not be what builds the
+        // cache, so an un-warmed client still reports zero entries.
+        let group_cache = match self.group_cache.get() {
             // Arc<T>'s HeapSize already includes size_of::<GroupInfo>().
             Some(cache) => {
                 cache
@@ -188,10 +183,14 @@ impl Client {
 
         // Each count read into a local so no two guards are ever held at once.
         let response_waiters = self.response_waiters_guard().len();
-        let presence_subscriptions = self.presence_subscriptions.lock().await.len();
+        let presence_subscriptions = self
+            .presence_subscriptions
+            .lock()
+            .unwrap_or_else(|p| p.into_inner())
+            .len();
         let app_state_key_requests = self.app_state_key_requests.lock().await.len();
         let app_state_syncing = self.app_state_syncing.len();
-        let chatstate_handlers = self.chatstate_handlers.read().await.len();
+        let chatstate_handlers = self.chatstate_handler_count.load(Ordering::Acquire);
         let history_sync_activity = self.history_sync_activity.snapshot();
         let history_sync_tasks = CollectionStats::new(
             history_sync_activity.tasks as u64,

src/client/adapters.rs

@@ -51,10 +51,10 @@ impl Client {
     }
 
     /// Get the active noise socket, or error if not connected.
-    pub(crate) async fn get_noise_socket(&self) -> Result<Arc<NoiseSocket>, ClientError> {
+    pub(crate) fn get_noise_socket(&self) -> Result<Arc<NoiseSocket>, ClientError> {
         self.noise_socket
             .lock()
-            .await
+            .unwrap_or_else(|p| p.into_inner())
             .clone()
             .ok_or(ClientError::NotConnected)
     }

src/client/app_state.rs

@@ -642,18 +642,14 @@ fn finalize_app_state_key_request_peers(
 }
 
 impl Client {
-    pub(crate) async fn get_app_state_processor(&self) -> Arc<AppStateProcessor> {
-        let mut guard = self.app_state_processor.lock().await;
-        if let Some(proc) = guard.as_ref() {
-            return proc.clone();
-        }
-        debug!("Initializing AppStateProcessor for the first time.");
-        let proc = Arc::new(AppStateProcessor::new(
-            self.persistence_manager.backend(),
-            self.runtime.clone(),
-        ));
-        *guard = Some(proc.clone());
-        proc
+    pub(crate) fn get_app_state_processor(&self) -> &Arc<AppStateProcessor> {
+        self.app_state_processor.get_or_init(|| {
+            debug!("Initializing AppStateProcessor for the first time.");
+            Arc::new(AppStateProcessor::new(
+                self.persistence_manager.backend(),
+                self.runtime.clone(),
+            ))
+        })
     }
 
     /// Pre-download every external blob (snapshots + patch external mutations)
@@ -1698,7 +1694,7 @@ impl Client {
                 });
             }
 
-            let proc = self.get_app_state_processor().await;
+            let proc = self.get_app_state_processor();
             // Pre-download all external blobs for all collections in the response,
             // concurrently (independent CDN GETs, keyed by directPath).
             let pre_downloaded = self.pre_download_external_blobs(&patch_lists).await;
@@ -2008,7 +2004,7 @@ impl Client {
             debug!(target: "Client/AppState", "Parsed patch list for {:?}: has_snapshot_ref={} has_more_patches={} patches_count={}",
                 name, pl.snapshot_ref.is_some(), pl.has_more_patches, pl.patches.len());
 
-            let proc = self.get_app_state_processor().await;
+            let proc = self.get_app_state_processor();
 
             // Pre-download all external blobs (snapshot and patch mutations),
             // concurrently, keyed by directPath.
@@ -2394,7 +2390,7 @@ impl Client {
             ),
             None => None,
         };
-        let proc = self.get_app_state_processor().await;
+        let proc = self.get_app_state_processor();
 
         for attempt in 1..=APP_STATE_PATCH_SEND_ATTEMPTS {
             // Cloned per attempt because a conflict rebuilds the patch against
@@ -2531,7 +2527,7 @@ impl Client {
                     .cloned()
                     .ok_or_else(|| anyhow::anyhow!("external blob not pre-downloaded: {path}"))
             };
-            let proc = self.get_app_state_processor().await;
+            let proc = self.get_app_state_processor();
             match proc.process_parsed_patch_list(list, &download, true).await {
                 Ok((mutations, _, _)) => {
                     wacore::telemetry::appstate_mutations(mutations.len() as u64);

src/client/device_registry.rs

@@ -3493,11 +3493,7 @@ mod tests {
 
         let hashed: Jid = format!("{contact_lid}@lid").parse().expect("jid");
         assert!(
-            client
-                .pending_device_sync
-                .take_all()
-                .await
-                .contains(&hashed),
+            client.pending_device_sync.take_all().contains(&hashed),
             "the hashed contact must be queued for a device-list refresh"
         );
     }
@@ -3522,7 +3518,7 @@ mod tests {
             .await;
 
         assert!(
-            client.pending_device_sync.take_all().await.is_empty(),
+            client.pending_device_sync.take_all().is_empty(),
             "an unresolvable hash must not refresh an unrelated contact"
         );
     }

src/client/lifecycle.rs

@@ -343,7 +343,7 @@ impl Client {
             transport: Arc::new(Mutex::new(None)),
             transport_events: Arc::new(Mutex::new(None)),
             transport_factory,
-            noise_socket: Arc::new(Mutex::new(None)),
+            noise_socket: Arc::new(std::sync::Mutex::new(None)),
 
             response_waiters: Arc::new(std::sync::Mutex::new(ResponseWaiterMap::default())),
             node_waiters: std::sync::Mutex::new(Vec::new()),
@@ -378,7 +378,7 @@ impl Client {
                 cache_config.cache_stores.lid_pn_cache.clone(),
             )),
             ab_props: Arc::new(wacore::store::ab_props::AbPropsCache::new()),
-            group_cache: Mutex::new(None),
+            group_cache: std::sync::OnceLock::new(),
 
             expected_disconnect: Arc::new(AtomicBool::new(false)),
             intentional_reconnect: AtomicBool::new(false),
@@ -421,7 +421,7 @@ impl Client {
 
             needs_initial_full_sync: Arc::new(app_state::BootstrapGate::new(false)),
 
-            app_state_processor: Mutex::new(None),
+            app_state_processor: std::sync::OnceLock::new(),
             app_state_key_requests: Arc::new(Mutex::new(HashMap::new())),
             app_state_syncing: app_state::SyncInFlight::new(),
             app_state_send_lock: Arc::new(Mutex::new(())),
@@ -438,7 +438,7 @@ impl Client {
             outbound_flush: Arc::new(crate::flush_scope::FlushScope::new()),
             delivery_receipt_queue: std::sync::OnceLock::new(),
             transport_ack_queue: std::sync::OnceLock::new(),
-            presence_subscriptions: Arc::new(Mutex::new(HashSet::new())),
+            presence_subscriptions: Arc::new(std::sync::Mutex::new(HashSet::new())),
             socket_ready_notifier: Arc::new(event_listener::Event::new()),
             is_ready: Arc::new(AtomicBool::new(false)),
             connected_notifier: Arc::new(event_listener::Event::new()),
@@ -460,10 +460,13 @@ impl Client {
             signal_flush_test_in_attempt: AtomicU32::new(0),
             #[cfg(test)]
             app_state_key_share_prepare_test_failures: AtomicU32::new(0),
+            #[cfg(test)]
+            chatstate_events_built: AtomicU32::new(0),
             custom_enc_handlers: std::sync::OnceLock::new(),
             inbound_durability_hook: std::sync::OnceLock::new(),
             retry_admission: std::sync::OnceLock::new(),
-            chatstate_handlers: Arc::new(RwLock::new(Vec::new())),
+            chatstate_handlers: Arc::new(std::sync::RwLock::new(Arc::from([]))),
+            chatstate_handler_count: AtomicUsize::new(0),
             pdo_pending_requests: cache_config.pdo_pending_requests.build_with_ttl(),
             pdo_requested: cache_config.pdo_requested.build_with_ttl(),
             device_registry_cache: device_topology::DeviceRegistryCache::new(
@@ -844,7 +847,7 @@ impl Client {
 
         *self.transport.lock().await = Some(transport);
         *self.transport_events.lock().await = Some(transport_events);
-        *self.noise_socket.lock().await = Some(noise_socket);
+        *self.noise_socket.lock().unwrap_or_else(|p| p.into_inner()) = Some(noise_socket);
         self.is_connected.store(true, Ordering::Release);
 
         // Notify waiters that socket is ready (before login)
@@ -1177,7 +1180,7 @@ impl Client {
         // afterwards would strip the replacement connection instead of the one being torn down.
         let transport = self.transport.lock().await.take();
         *self.transport_events.lock().await = None;
-        *self.noise_socket.lock().await = None;
+        *self.noise_socket.lock().unwrap_or_else(|p| p.into_inner()) = None;
         if let Some(transport) = transport {
             transport.disconnect().await;
         }
@@ -1235,7 +1238,7 @@ impl Client {
         // Reset dead-socket timestamps so stale values from the previous
         // connection don't trigger an immediate reconnect on the next one.
         self.stats.reset_connection_activity();
-        self.pending_device_sync.clear().await;
+        self.pending_device_sync.clear();
         // Reset offline sync state for next connection
         self.offline_sync_completed.store(false, Ordering::Relaxed);
         self.offline_sync_finish_started
@@ -1297,9 +1300,11 @@ impl Client {
         *self.media_conn.write().await = None;
 
         // Clear app state key cache — keys will be re-fetched from DB on demand
-        let processor = self.app_state_processor.lock().await.clone();
-        if let Some(processor) = processor {
-            processor.clear_key_cache().await;
+        // main took the processor out of the mutex before awaiting so the guard
+        // did not span the clear; the write-once cell has no guard to span, so
+        // the borrow is the whole of it.
+        if let Some(proc) = self.app_state_processor.get() {
+            proc.clear_key_cache().await;
         }
         #[cfg(feature = "client-lifecycle")]
         drop(scope_close);

src/client/messaging.rs

@@ -12,7 +12,7 @@ impl Client {
     /// This bypasses node logging and `sent_node_waiter` resolution — use
     /// [`send_node`](Client::send_node) for normal stanza sending.
     pub async fn send_raw_bytes(&self, plaintext: Vec<u8>) -> Result<(), ClientError> {
-        let noise_socket = self.get_noise_socket().await?;
+        let noise_socket = self.get_noise_socket()?;
         // Wire bytes and the last-sent timestamp are recorded by the noise
         // sender task at the actual transport write.
         noise_socket
@@ -53,7 +53,7 @@ impl Client {
         results: &mut Vec<crate::socket::error::EncryptSendResult>,
     ) -> Result<(), ClientError> {
         results.clear();
-        let noise_socket = match self.get_noise_socket().await {
+        let noise_socket = match self.get_noise_socket() {
             Ok(socket) => socket,
             Err(error) => {
                 frames.clear();
@@ -455,11 +455,19 @@ impl Client {
     /// Register a chatstate handler which will be invoked when a `<chatstate>` stanza is received.
     ///
     /// The handler receives a `ChatStateEvent` with the parsed chat state information.
-    pub async fn register_chatstate_handler(
-        &self,
-        handler: Arc<dyn Fn(ChatStateEvent) + Send + Sync>,
-    ) {
-        self.chatstate_handlers.write().await.push(handler);
+    pub fn register_chatstate_handler(&self, handler: Arc<dyn Fn(ChatStateEvent) + Send + Sync>) {
+        let mut guard = self
+            .chatstate_handlers
+            .write()
+            .unwrap_or_else(|p| p.into_inner());
+        let mut handlers = Vec::with_capacity(guard.len() + 1);
+        handlers.extend(guard.iter().cloned());
+        handlers.push(handler);
+        *guard = Arc::from(handlers);
+        // Published after the snapshot is in place, so a reader that sees a
+        // non-zero count always finds the handler behind it.
+        self.chatstate_handler_count
+            .store(guard.len(), Ordering::Release);
     }
 
     /// Dispatch a parsed chatstate stanza to registered handlers.
@@ -512,10 +520,20 @@ impl Client {
                 .build(),
         ));
 
-        // Invoke legacy callback handlers
+        // Invoke legacy callback handlers. Building the event is only worth it
+        // once something reads it, and the default registers nothing.
+        if self.chatstate_handler_count.load(Ordering::Acquire) == 0 {
+            return;
+        }
+        #[cfg(test)]
+        self.chatstate_events_built.fetch_add(1, Ordering::Release);
         let event = ChatStateEvent::from_stanza(stanza);
-        let handlers = self.chatstate_handlers.read().await.clone();
-        for handler in handlers {
+        let handlers = self
+            .chatstate_handlers
+            .read()
+            .unwrap_or_else(|p| p.into_inner())
+            .clone();
+        for handler in handlers.iter().cloned() {
             let event_clone = event.clone();
             self.runtime
                 .spawn(Box::pin(async move {

src/client/node_io.rs

@@ -163,7 +163,6 @@ impl Client {
         // so resolve it once instead of locking the mutex per frame.
         let noise_socket = self
             .get_noise_socket()
-            .await
             .map_err(|_| ReadLoopError::NotStarted("no noise socket"))?;
 
         // Frame decoder to parse incoming data