perf: parallelize remaining serial/blocking hot paths (startup, media, send/recv)

src/bot.rs

@@ -467,7 +467,19 @@ impl Bot {
         } = self;
 
         if let Some(receiver) = sync_task_receiver {
+            // History-sync tasks are independent — they upsert message secrets /
+            // tctokens (order-free) and the dispatched event carries chunk_order
+            // for consumers — so they can ingest concurrently. App-state tasks stay
+            // strictly serial and inline: same-collection patch application is
+            // order-sensitive (LTHash). The history cap is deliberately low because
+            // each task transiently holds a decompressed history blob and the
+            // connect path is peak-memory-conscious; WA Web caps history-sync
+            // chunks at histSyncChunk=3, we stay at 2 for headroom. The permit is
+            // taken in the recv loop so a burst applies backpressure instead of
+            // spawning unbounded tasks that each pin a compressed blob.
+            const HISTORY_SYNC_CONCURRENCY: usize = 2;
             let worker_client = Arc::downgrade(&client);
+            let history_permits = Arc::new(async_lock::Semaphore::new(HISTORY_SYNC_CONCURRENCY));
             client
                 .runtime
                 .spawn(Box::pin(async move {
@@ -476,7 +488,20 @@ impl Bot {
                             break;
                         };
 
-                        worker_client.process_sync_task(task).await;
+                        if matches!(task, crate::sync_task::MajorSyncTask::HistorySync { .. }) {
+                            let permit = history_permits.acquire_arc().await;
+                            let task_client = worker_client.clone();
+                            worker_client
+                                .runtime
+                                .spawn(Box::pin(async move {
+                                    let _permit = permit;
+                                    task_client.process_sync_task(task).await;
+                                }))
+                                .detach();
+                        } else {
+                            // App-state sync: serial + ordered.
+                            worker_client.process_sync_task(task).await;
+                        }
                     }
                     info!("Sync worker shutting down.");
                 }))

src/client/app_state.rs

@@ -2,6 +2,17 @@
 
 use super::*;
 
+/// Concurrency cap for pre-downloading app-state external blobs. Each blob is an
+/// independent CDN GET (a collection snapshot or a patch's external mutations),
+/// keyed by directPath, so they carry no ordering dependency (LTHash ordering is
+/// in patch *application*, not blob *fetching*). WA Web downloads them in
+/// parallel — `Syncd/CollectionHandler` fans the per-patch external-mutation
+/// fetches out under `Promise.all`. Bounded (not unbounded `join_all`) because a
+/// snapshot can be multi-MB and a batched response carries several collections;
+/// the cap keeps peak memory in check while still turning `sum(RTT)` into
+/// `~max(RTT)`.
+const APPSTATE_BLOB_DOWNLOAD_CONCURRENCY: usize = 4;
+
 impl Client {
     pub(crate) async fn get_app_state_processor(&self) -> Arc<AppStateProcessor> {
         let mut guard = self.app_state_processor.lock().await;
@@ -17,6 +28,97 @@ impl Client {
         proc
     }
 
+    /// Pre-download every external blob (collection snapshots + patch external
+    /// mutations) referenced by `patch_lists`, keyed by directPath. The blobs are
+    /// independent, so the CDN GETs run concurrently (bounded by
+    /// [`APPSTATE_BLOB_DOWNLOAD_CONCURRENCY`]) instead of one RTT at a time. A
+    /// failed download is logged and omitted from the map; the later inline step
+    /// (`missing_key_ids_after_inline` / `process_patch_lists`) surfaces the
+    /// missing blob exactly as the previous serial version did. Mirrors WA Web's
+    /// parallel syncd blob fetch (`Syncd/CollectionHandler` `Promise.all`).
+    async fn pre_download_external_blobs(
+        &self,
+        patch_lists: &[wacore::appstate::patch_decode::PatchList],
+    ) -> std::collections::HashMap<String, Vec<u8>> {
+        use futures::StreamExt;
+
+        // Blob context, kept only so a failed download logs the same message the
+        // serial version did (snapshot name vs. patch version).
+        enum BlobKind {
+            Snapshot(WAPatchName),
+            Mutation(u64),
+        }
+
+        // Collect every (blob, context) to fetch. The blob reference is cloned
+        // (it's small: a directPath + key/hash bytes) so each concurrent task owns
+        // its input and captures only `&self` — mirrors the owned-data fan-out in
+        // `groups::fill_participant_pns` and keeps the future `Send`. Only blobs
+        // with a directPath are enqueued; the directPath is recovered from the
+        // moved `ext` after the fetch (no separate clone for the map key).
+        let mut jobs: Vec<(wa::ExternalBlobReference, BlobKind)> = Vec::new();
+        for pl in patch_lists {
+            if let Some(ext) = &pl.snapshot_ref
+                && ext.direct_path.is_some()
+            {
+                jobs.push((ext.clone(), BlobKind::Snapshot(pl.name)));
+            }
+            for patch in &pl.patches {
+                if let Some(ext) = patch.external_mutations.as_option()
+                    && ext.direct_path.is_some()
+                {
+                    let v = patch
+                        .version
+                        .as_option()
+                        .and_then(|v| v.version)
+                        .unwrap_or(0);
+                    jobs.push((ext.clone(), BlobKind::Mutation(v)));
+                }
+            }
+        }
+
+        if jobs.is_empty() {
+            return std::collections::HashMap::new();
+        }
+
+        let mut pre_downloaded = std::collections::HashMap::with_capacity(jobs.len());
+        let results = futures::stream::iter(jobs.into_iter().map(|(ext, kind)| async move {
+            let bytes = self.download(&ext).await;
+            // directPath presence was checked when the job was built.
+            (ext.direct_path, kind, bytes)
+        }))
+        .buffer_unordered(APPSTATE_BLOB_DOWNLOAD_CONCURRENCY)
+        .collect::<Vec<_>>()
+        .await;
+
+        for (path, kind, res) in results {
+            match res {
+                Ok(bytes) => {
+                    if let BlobKind::Mutation(v) = kind {
+                        debug!(target: "Client/AppState", "Downloaded external mutations for patch v{} ({} bytes)", v, bytes.len());
+                    } else {
+                        debug!(target: "Client/AppState", "Downloaded external snapshot ({} bytes)", bytes.len());
+                    }
+                    if let Some(path) = path {
+                        pre_downloaded.insert(path, bytes);
+                    }
+                }
+                Err(e) => match kind {
+                    BlobKind::Snapshot(name) => {
+                        warn!("Failed to download external snapshot for {:?}: {e}", name)
+                    }
+                    BlobKind::Mutation(v) => {
+                        warn!(
+                            "Failed to download external mutations for patch v{}: {e}",
+                            v
+                        )
+                    }
+                },
+            }
+        }
+
+        pre_downloaded
+    }
+
     /// Public entry point for processing [`MajorSyncTask`] from the sync channel.
     #[cfg_attr(
         feature = "tracing",
@@ -218,60 +320,15 @@ impl Client {
 
             let resp = self.send_iq(iq).await?;
 
-            // Pre-download all external blobs for all collections in the response
-            let mut pre_downloaded: std::collections::HashMap<String, Vec<u8>> =
-                std::collections::HashMap::new();
-
             // Parse the response once here for pre-download; the same parsed
             // lists are handed to the processor below (no second parse).
             let mut patch_lists =
                 wacore::appstate::patch_decode::parse_patch_lists_ref(resp.get())?;
 
             let proc = self.get_app_state_processor().await;
-            {
-                for pl in &patch_lists {
-                    // Download external snapshot
-                    if let Some(ext) = &pl.snapshot_ref
-                        && let Some(path) = &ext.direct_path
-                    {
-                        match self.download(ext).await {
-                            Ok(bytes) => {
-                                pre_downloaded.insert(path.clone(), bytes);
-                            }
-                            Err(e) => {
-                                warn!(
-                                    "Failed to download external snapshot for {:?}: {e}",
-                                    pl.name
-                                );
-                            }
-                        }
-                    }
-
-                    // Download external mutations
-                    for patch in &pl.patches {
-                        if let Some(ext) = patch.external_mutations.as_option()
-                            && let Some(path) = &ext.direct_path
-                        {
-                            match self.download(ext).await {
-                                Ok(bytes) => {
-                                    pre_downloaded.insert(path.clone(), bytes);
-                                }
-                                Err(e) => {
-                                    let v = patch
-                                        .version
-                                        .as_option()
-                                        .and_then(|v| v.version)
-                                        .unwrap_or(0);
-                                    warn!(
-                                        "Failed to download external mutations for patch v{}: {e}",
-                                        v
-                                    );
-                                }
-                            }
-                        }
-                    }
-                }
-            }
+            // 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;
 
             let download = |ext: &wa::ExternalBlobReference| -> anyhow::Result<Vec<u8>> {
                 if let Some(path) = &ext.direct_path {
@@ -477,51 +534,11 @@ impl Client {
 
             let proc = self.get_app_state_processor().await;
 
-            // Pre-download all external blobs (snapshot and patch mutations); keyed by
-            // directPath.
-            let mut pre_downloaded: std::collections::HashMap<String, Vec<u8>> =
-                std::collections::HashMap::new();
-            {
-                // Download external snapshot if present
-                if let Some(ext) = &pl.snapshot_ref
-                    && let Some(path) = &ext.direct_path
-                {
-                    match self.download(ext).await {
-                        Ok(bytes) => {
-                            debug!(target: "Client/AppState", "Downloaded external snapshot ({} bytes)", bytes.len());
-                            pre_downloaded.insert(path.clone(), bytes);
-                        }
-                        Err(e) => {
-                            warn!("Failed to download external snapshot: {e}");
-                        }
-                    }
-                }
-
-                // Download external mutations for each patch that has them
-                for patch in &pl.patches {
-                    if let Some(ext) = patch.external_mutations.as_option()
-                        && let Some(path) = &ext.direct_path
-                    {
-                        let patch_version = patch
-                            .version
-                            .as_option()
-                            .and_then(|v| v.version)
-                            .unwrap_or(0);
-                        match self.download(ext).await {
-                            Ok(bytes) => {
-                                debug!(target: "Client/AppState", "Downloaded external mutations for patch v{} ({} bytes)", patch_version, bytes.len());
-                                pre_downloaded.insert(path.clone(), bytes);
-                            }
-                            Err(e) => {
-                                warn!(
-                                    "Failed to download external mutations for patch v{}: {e}",
-                                    patch_version
-                                );
-                            }
-                        }
-                    }
-                }
-            }
+            // Pre-download all external blobs (snapshot and patch mutations),
+            // concurrently, keyed by directPath.
+            let pre_downloaded = self
+                .pre_download_external_blobs(std::slice::from_ref(&pl))
+                .await;
 
             let download = |ext: &wa::ExternalBlobReference| -> anyhow::Result<Vec<u8>> {
                 if let Some(path) = &ext.direct_path {

src/client/node_io.rs

@@ -743,11 +743,33 @@ impl Client {
                 debug!("Skipping passive tasks: connection closed");
                 return;
             }
-            if let Err(e) = client_clone.upload_pre_keys_at_login().await
-                && !client_clone.is_shutting_down()
-            {
-                warn!("Failed to upload pre-keys during startup: {e:?}");
-            }
+            // WA Web PassiveTasks: the one-time pre-key upload is a passive task
+            // (`PassiveTaskManager.registerPassiveTask("KeyUpload", ...)`), not a
+            // gate on going active. Uploading publishes pre-keys for peers' FUTURE
+            // sessions; decrypting the offline backlog only needs pre-keys we
+            // already hold locally, and on a fresh device the server has none yet
+            // so no incoming pkmsg can reference them. Awaiting it here only
+            // delayed offline delivery, so spawn it like RotateKeyJob below.
+            // No-op on reconnect via the persisted server_has_prekeys flag.
+            check_generation!();
+            let prekey_client = client_clone.clone();
+            let prekey_generation = task_generation;
+            client_clone
+                .runtime
+                .spawn(Box::pin(async move {
+                    // A newer connection may have taken over between spawn and now.
+                    if prekey_client.connection_generation.load(Ordering::SeqCst)
+                        != prekey_generation
+                    {
+                        return;
+                    }
+                    if let Err(e) = prekey_client.upload_pre_keys_at_login().await
+                        && !prekey_client.is_shutting_down()
+                    {
+                        warn!("Failed to upload pre-keys during startup: {e:?}");
+                    }
+                }))
+                .detach();
 
             // WA Web RotateKeyJob: rotate the signed pre-key on its cadence.
             // Spawned so a slow or failing encrypt IQ never delays the rest of

src/client/sessions.rs

@@ -288,21 +288,34 @@ impl Client {
         use wacore::types::jid::JidExt;
 
         let device_snapshot = self.persistence_manager.get_device_snapshot();
-        let mut jids_needing_sessions = Vec::with_capacity(jids.len());
 
-        for jid in jids {
-            let signal_addr = jid.to_protocol_address();
-            // Check cache first (includes unflushed sessions), fall back to backend
-            match self
-                .signal_cache
-                .has_session(&signal_addr, &*device_snapshot.backend)
-                .await
-            {
-                Ok(true) => {}
-                Ok(false) => jids_needing_sessions.push(jid),
-                Err(e) => log::warn!("Failed to check session for {}: {}", jid.observe(), e),
-            }
-        }
+        // Check each session concurrently. Warm hits serialize on the signal
+        // cache's mutex regardless, but a cold-cache multi-recipient ensure (e.g.
+        // status / group session setup over many devices) would otherwise
+        // serialize the per-device DB reads. Each probe is independent and the
+        // resulting order is irrelevant (the misses are chunked for the fetch).
+        use futures::StreamExt;
+        let backend = device_snapshot.backend.clone();
+        let jids_needing_sessions: Vec<Jid> = futures::stream::iter(jids)
+            .map(|jid| {
+                let backend = backend.clone();
+                async move {
+                    let signal_addr = jid.to_protocol_address();
+                    // Check cache first (includes unflushed sessions), fall back to backend.
+                    match self.signal_cache.has_session(&signal_addr, &*backend).await {
+                        Ok(true) => None,
+                        Ok(false) => Some(jid),
+                        Err(e) => {
+                            log::warn!("Failed to check session for {}: {}", jid.observe(), e);
+                            None
+                        }
+                    }
+                }
+            })
+            .buffer_unordered(16)
+            .filter_map(|needed| async move { needed })
+            .collect()
+            .await;
 
         if jids_needing_sessions.is_empty() {
             return Ok(());