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

src/bot.rs

@@ -467,7 +467,15 @@ impl Bot {
         } = self;
 
         if let Some(receiver) = sync_task_receiver {
+            // History-sync chunks are independent (order-free upserts; the event
+            // carries chunk_order), so ingest them concurrently. Bounded low: each
+            // holds a decompressed blob and the connect path is peak-memory-
+            // conscious (WA Web caps at histSyncChunk=3). App-state stays serial
+            // (order-sensitive patch application). The permit is taken in the recv
+            // loop so a burst backpressures instead of piling up blob-pinning tasks.
+            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,9 +484,24 @@ 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.");
+                    info!(
+                        "Sync worker intake loop finished (detached history-sync tasks may still be running)."
+                    );
                 }))
                 .detach();
         }

src/client/app_state.rs

@@ -2,6 +2,12 @@
 
 use super::*;
 
+/// Concurrency cap for pre-downloading app-state external blobs (independent CDN
+/// GETs, keyed by directPath — LTHash ordering is in patch application, not blob
+/// fetching). WA Web fans these out under `Promise.all` (`Syncd/CollectionHandler`);
+/// bounded here because a snapshot can be multi-MB and a batch carries several.
+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 +23,94 @@ impl Client {
         proc
     }
 
+    /// Pre-download every external blob (snapshots + patch external mutations)
+    /// referenced by `patch_lists`, keyed by directPath, fetching concurrently
+    /// (bounded by [`APPSTATE_BLOB_DOWNLOAD_CONCURRENCY`]). A failed download is
+    /// logged and omitted; the later inline step surfaces the missing blob as
+    /// before. Mirrors WA Web's parallel syncd blob fetch.
+    async fn pre_download_external_blobs(
+        &self,
+        patch_lists: &[wacore::appstate::patch_decode::PatchList],
+    ) -> std::collections::HashMap<String, Vec<u8>> {
+        use futures::StreamExt;
+
+        // Kept only so a failed download logs the right message (snapshot vs patch).
+        enum BlobKind {
+            Snapshot(WAPatchName),
+            Mutation(u64),
+        }
+
+        // Clone the (small) blob ref into each job so the task owns its input and
+        // captures only `&self` (keeps the future Send); the directPath is
+        // recovered from the moved `ext` after the fetch. Dedup by directPath so
+        // patches sharing a blob don't fetch it twice into the same map key.
+        let mut jobs: Vec<(wa::ExternalBlobReference, BlobKind)> = Vec::new();
+        let mut seen_paths: std::collections::HashSet<&str> = std::collections::HashSet::new();
+        for pl in patch_lists {
+            if let Some(ext) = &pl.snapshot_ref
+                && let Some(path) = ext.direct_path.as_deref()
+                && seen_paths.insert(path)
+            {
+                jobs.push((ext.clone(), BlobKind::Snapshot(pl.name)));
+            }
+            for patch in &pl.patches {
+                if let Some(ext) = patch.external_mutations.as_option()
+                    && let Some(path) = ext.direct_path.as_deref()
+                    && seen_paths.insert(path)
+                {
+                    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 +312,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 +526,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,30 @@ 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 pre-key upload is a passive task, not a gate
+            // on going active — it only publishes keys for peers' FUTURE sessions
+            // (the offline backlog uses keys we already hold, and a fresh device's
+            // server pool is empty). Awaiting it here just delayed offline delivery,
+            // so spawn it like RotateKeyJob below.
+            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,32 @@ 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),
-            }
-        }
+        // Probe sessions concurrently: a cold-cache multi-recipient ensure would
+        // otherwise serialize the per-device DB reads (warm hits serialize on the
+        // cache mutex anyway). Order is irrelevant — 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(());