From 05641bd9877097b2f030e44a632823b2848ae9a0 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:36 +0000 Subject: [PATCH 01/30] perf(appstate): download external blobs concurrently during sync The app-state sync pre-download step fetched every external blob (collection snapshots + per-patch external mutations) with one awaited CDN GET at a time, so initial-sync latency was the sum of every blob's round-trip. The blobs are independent (keyed by directPath; LTHash ordering lives in patch *application*, not blob *fetching*), so fetch them concurrently behind a bounded window. Mirrors WA Web, which fans the per-patch external-mutation downloads out under `Promise.all` in `Syncd/CollectionHandler`. Bounded (not unbounded) because a snapshot can be multi-MB and a batched response carries several collections; the cap keeps peak memory in check while turning sum(RTT) into ~max(RTT). Shared helper covers both the batched (`sync_collections_batched`) and single-collection (`process_app_state_sync_task`) paths. Each task owns its (cloned) blob reference and captures only `&self`, keeping the future Send. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0116cfRpumbEv3H4qUNmtNk4 --- src/client/app_state.rs | 203 ++++++++++++++++++++++------------------ 1 file changed, 110 insertions(+), 93 deletions(-) diff --git a/src/client/app_state.rs b/src/client/app_state.rs index da7c8a9f4..cc6daf698 100644 --- a/src/client/app_state.rs +++ b/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 { 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> { + 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::>() + .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> = - 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> { 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> = - 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> { if let Some(path) = &ext.direct_path { From b8f28f4a777d28386e75add33b25c2cbe7cfa2dc Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:36 +0000 Subject: [PATCH 02/30] perf(media): stream buffered download() instead of fetch-then-decrypt MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `download()` fetched the entire ciphertext into a Vec on one blocking hop and then decrypted it on a second — so wall time was download + decrypt with the full ciphertext and plaintext resident at once. Route it through the existing streaming writer path (into an in-memory buffer): the CDN read and the AES/HMAC decrypt interleave in a single blocking pass, overlapping network with CPU and roughly halving peak memory (no separate full-ciphertext buffer). The streaming path already falls back to a buffered fetch+decrypt for non-streaming HTTP clients, so behavior is strictly >= before. WA Web is the same shape: it overlaps key derivation with the fetch under Promise.all and offloads decrypt to a worker. Retry safety: every host serves the same blob decrypting to the same length, and status is checked before any write on auth/not-found errors, so a retry that reuses the buffer always rewrites at least the partially-written bytes — no stale tail. The buffer is pre-sized to the declared plaintext length (capped) so the common case is a single allocation. Removes the now-dead buffered helpers (`download_media_with_retry`, `download_with_request`); their auth-refresh + host-failover retry behavior is covered by the surviving `download_to_writer` retry test. Co-Authored-By: Claude Opus 4.8 Claude-Session: https://claude.ai/code/session_0116cfRpumbEv3H4qUNmtNk4 --- src/download.rs | 218 +++++++----------------------------------------- 1 file changed, 30 insertions(+), 188 deletions(-) diff --git a/src/download.rs b/src/download.rs index 6ca633518..82b38cc16 100644 --- a/src/download.rs +++ b/src/download.rs @@ -7,6 +7,13 @@ pub use wacore::download::{ DownloadUtils, Downloadable, MediaDecryption, MediaDecryptionError, MediaType, }; +/// Cap on the speculative capacity pre-allocated for the in-memory download +/// buffer. Sized to the plaintext length the message declares, but a bogus +/// length must not drive a multi-GB allocation before a single byte arrives; +/// beyond this the buffer grows on demand. Comfortably above typical +/// image/video/audio media so the common case is a single allocation. +const DOWNLOAD_PREALLOC_CAP: u64 = 64 * 1024 * 1024; + impl From<&MediaConn> for wacore::download::MediaConnection { fn from(conn: &MediaConn) -> Self { wacore::download::MediaConnection { @@ -116,70 +123,6 @@ impl DownloadRequestError { } } -async fn download_media_with_retry< - PrepareRequests, - PrepareRequestsFut, - InvalidateMediaConn, - InvalidateMediaConnFut, - ExecuteRequest, - ExecuteRequestFut, ->( - mut prepare_requests: PrepareRequests, - mut invalidate_media_conn: InvalidateMediaConn, - mut execute_request: ExecuteRequest, -) -> Result> -where - PrepareRequests: FnMut(bool) -> PrepareRequestsFut, - PrepareRequestsFut: - std::future::Future>>, - InvalidateMediaConn: FnMut() -> InvalidateMediaConnFut, - InvalidateMediaConnFut: std::future::Future, - ExecuteRequest: FnMut(wacore::download::DownloadRequest) -> ExecuteRequestFut, - ExecuteRequestFut: - std::future::Future, DownloadRequestError>>, -{ - let mut force_refresh = false; - let mut last_err: Option = None; - - for attempt in 0..=MEDIA_AUTH_REFRESH_RETRY_ATTEMPTS { - let requests = prepare_requests(force_refresh).await?; - let mut retry_with_fresh_auth = false; - - for request in requests { - match execute_request(request.clone()).await { - Ok(data) => return Ok(data), - Err(err) if (err.is_auth() || err.is_not_found()) && attempt == 0 => { - // Auth error or 404/410 (expired URL): refresh media conn and re-derive URLs. - // Matches WA Web's MediaNotFoundError → forceRefresh flow. - invalidate_media_conn().await; - force_refresh = true; - retry_with_fresh_auth = true; - break; - } - Err(err) if err.is_auth() || err.is_not_found() => return Err(err.into_anyhow()), - Err(err) => { - let err = err.into_anyhow(); - log::warn!( - "Failed to download from URL {}: {:?}. Trying next host.", - request.url, - err - ); - last_err = Some(err); - } - } - } - - if !retry_with_fresh_auth { - break; - } - } - - match last_err { - Some(err) => Err(err), - None => Err(anyhow!("Failed to download from all available media hosts")), - } -} - async fn download_to_writer_with_retry< W, PrepareRequests, @@ -259,12 +202,25 @@ impl Client { tracing::instrument(name = "wa.media.download", level = "debug", skip_all, err(Debug)) )] pub async fn download(&self, downloadable: &dyn Downloadable) -> Result> { - download_media_with_retry( - |force| self.prepare_requests(downloadable, force), - || async { self.invalidate_media_conn().await }, - |request| async move { self.download_with_request(&request).await }, - ) - .await + // Route through the streaming writer path (into an in-memory buffer) + // instead of "fetch the whole ciphertext, then decrypt it in a second + // pass". The streaming path interleaves the CDN read with the AES/HMAC + // decrypt in a single blocking pass, overlapping network with CPU and + // roughly halving peak memory (no full-ciphertext buffer resident + // alongside the plaintext). `download_to_writer` falls back to a buffered + // fetch+decrypt automatically for non-streaming HTTP clients, so this is + // strictly >= the previous behavior. Pre-size the buffer to the plaintext + // length (the decrypted output size) when known, capped so a bogus + // server-declared length can't drive a huge speculative allocation. + let cap = downloadable + .file_length() + .unwrap_or(0) + .min(DOWNLOAD_PREALLOC_CAP) as usize; + let writer = std::io::Cursor::new(Vec::with_capacity(cap)); + Ok(self + .download_to_writer(downloadable, writer) + .await? + .into_inner()) } /// Fetch a first-party sticker pack's metadata and sticker list from the CDN. @@ -317,53 +273,6 @@ impl Client { DownloadUtils::prepare_download_requests(downloadable, &core_media_conn) } - async fn download_with_request( - &self, - request: &wacore::download::DownloadRequest, - ) -> std::result::Result, DownloadRequestError> { - let url = request.url.clone(); - let decryption = request.decryption.clone(); - let http_request = crate::http::HttpRequest::get(url); - let response = self - .http_client - .execute(http_request) - .await - .map_err(DownloadRequestError::other)?; - - if response.status_code >= 300 { - return Err(if is_media_auth_error(response.status_code) { - DownloadRequestError::auth(response.status_code) - } else if matches!(response.status_code, 404 | 410) { - DownloadRequestError::not_found(response.status_code) - } else { - DownloadRequestError::other(anyhow!( - "Download failed with status: {}", - response.status_code - )) - }); - } - - match decryption { - MediaDecryption::Encrypted { - media_key, - media_type, - } => wacore::runtime::blocking(&*self.runtime, move || { - DownloadUtils::decrypt_stream(&response.body[..], &media_key, media_type) - }) - .await - .map_err(DownloadRequestError::other), - MediaDecryption::Plaintext { file_sha256 } => { - let body = response.body; - wacore::runtime::blocking(&*self.runtime, move || { - DownloadUtils::validate_plaintext_sha256(&body, &file_sha256)?; - Ok::, anyhow::Error>(body) - }) - .await - .map_err(DownloadRequestError::other) - } - } - } - /// Downloads and decrypts media with streaming (constant memory usage). /// /// The entire HTTP download, decryption, and file write happen in a single @@ -646,77 +555,10 @@ mod tests { ); } - #[tokio::test] - async fn download_retries_with_forced_media_conn_refresh_after_auth_error() { - let body = b"download me".to_vec(); - let downloadable = PlaintextDownloadable { - direct_path: "/v/t62.7118-24/123".to_string(), - file_sha256: plaintext_sha256(&body), - }; - let first_conn = media_conn("stale-auth", &["cdn1.example.com"]); - let refreshed_conn = media_conn("fresh-auth", &["cdn2.example.com"]); - let refresh_calls = Arc::new(Mutex::new(Vec::new())); - let invalidations = Arc::new(Mutex::new(0usize)); - let seen_urls = Arc::new(Mutex::new(Vec::new())); - - let downloaded = download_media_with_retry( - { - let refresh_calls = Arc::clone(&refresh_calls); - let downloadable = &downloadable; - move |force| { - let refresh_calls = Arc::clone(&refresh_calls); - let first_conn = first_conn.clone(); - let refreshed_conn = refreshed_conn.clone(); - async move { - refresh_calls.lock().await.push(force); - let media_conn = if force { refreshed_conn } else { first_conn }; - DownloadUtils::prepare_download_requests( - downloadable, - &wacore::download::MediaConnection::from(&media_conn), - ) - } - } - }, - { - let invalidations = Arc::clone(&invalidations); - move || { - let invalidations = Arc::clone(&invalidations); - async move { - *invalidations.lock().await += 1; - } - } - }, - { - let seen_urls = Arc::clone(&seen_urls); - let body = body.clone(); - move |request| { - let seen_urls = Arc::clone(&seen_urls); - let body = body.clone(); - let url = request.url.clone(); - async move { - seen_urls.lock().await.push(url.clone()); - if url.contains("stale-auth") { - Err(DownloadRequestError::auth(401)) - } else { - Ok(body) - } - } - } - }, - ) - .await - .expect("download should succeed after refreshing media auth"); - - assert_eq!(downloaded, body); - assert_eq!(*refresh_calls.lock().await, vec![false, true]); - assert_eq!(*invalidations.lock().await, 1); - - let seen_urls = seen_urls.lock().await.clone(); - assert_eq!(seen_urls.len(), 2); - assert!(seen_urls[0].contains("auth=stale-auth")); - assert!(seen_urls[1].contains("auth=fresh-auth")); - } - + // The buffered `download()` now routes through `download_to_writer`, so its + // auth-refresh + host-failover retry behavior is exercised by + // `download_to_writer_retries_with_forced_media_conn_refresh_after_auth_error` + // below (same `download_to_writer_with_retry` engine). #[tokio::test] async fn download_to_writer_retries_with_forced_media_conn_refresh_after_auth_error() { let body = b"stream me".to_vec(); From 6e74ff58de82a4289e9a3e494fa0c3440a184162 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:36 +0000 Subject: [PATCH 03/30] perf(usync): resolve device lists from the registry concurrently MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit get_user_devices already batches the network usync into one IQ, but the local read scan that precedes it queried the device registry (in-memory cache, DB fallback) one user at a time. A cold-cache send to a large group (256+ members) serialized 256 registry reads — and 256 SQLite reads on an L1 miss — before the fetch IQ went out. Each read takes &self, is independent, and the resulting device-set order is irrelevant (the phash sorts and the encrypt fan-out is order-agnostic), so resolve them behind a bounded fan-out (16, matching the send encrypt fan-out and groups::fill_participant_pns). The network fetch of the still-missing users is unchanged (still one batched IQ). --- src/usync.rs | 42 +++++++++++++++++++++++++++++++++--------- 1 file changed, 33 insertions(+), 9 deletions(-) diff --git a/src/usync.rs b/src/usync.rs index d6a454511..1a2a4bd16 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -14,16 +14,40 @@ impl Client { let mut jids_to_fetch: HashSet = HashSet::with_capacity(jids.len()); let mut all_devices = Vec::with_capacity(jids.len() * 2); - for jid in jids.iter().map(|j| j.to_non_ad()) { - // Device registry (in-memory cache + DB) is the single source of truth. - // get_devices_from_registry returns None for an empty record (never a - // valid set — WA Web always keeps device 0), so a corrupted empty row - // falls through to the network here instead of being trusted. - if let Some(devices) = self.get_devices_from_registry(&jid).await { - all_devices.extend(devices); - continue; + // Resolve each user's device list from the registry concurrently. The + // network usync is already batched into one IQ below; this is the LOCAL + // read scan (in-memory cache, DB fallback per user) that a large group + // would otherwise serialize over 256+ members on a cold cache. Each read + // takes `&self` and is independent, and the resulting device set order is + // irrelevant (the phash sorts, and the encrypt fan-out is order-agnostic), + // so a bounded fan-out is safe. Bound (16) matches the send encrypt + // fan-out and `groups::fill_participant_pns`. + // + // get_devices_from_registry returns None for an empty record (never a + // valid set — WA Web always keeps device 0), so a corrupted empty row + // falls through to the network below instead of being trusted. + use futures::StreamExt; + // Materialize the non-AD users into an owned Vec so the stream owns its + // items (borrowing `jids` through `buffer_unordered` trips the future's + // higher-ranked Send bound — same reason `groups::fill_participant_pns` + // collects `pending` first). + let non_ad: Vec = jids.iter().map(|j| j.to_non_ad()).collect(); + let resolved: Vec<(Jid, Option>)> = futures::stream::iter(non_ad) + .map(|jid| async move { + let devices = self.get_devices_from_registry(&jid).await; + (jid, devices) + }) + .buffer_unordered(16) + .collect() + .await; + + for (jid, devices) in resolved { + match devices { + Some(devices) => all_devices.extend(devices), + None => { + jids_to_fetch.insert(jid); + } } - jids_to_fetch.insert(jid); } if !jids_to_fetch.is_empty() { From c3683caa67700f9b9498d50a24c1e629bf8be280 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:36 +0000 Subject: [PATCH 04/30] perf(recv): release the per-sender session lock before dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit process_session_enc_batch held the per-sender Signal session lock across the whole batch, including handle_decrypted_plaintext — message decode, SKDM / key-share handling, the durable inbound commit, the app-controlled durability hook, and synchronous event dispatch. None of that touches the pairwise ratchet the lock guards (PASS 2's group path already runs handle_decrypted_plaintext with no session lock at all), so holding the lock across it needlessly blocks concurrent same-sender work — a retry-receipt session rebuild, a reg-id / base-key session reset. Collect the successfully decrypted plaintexts during the (still lock-held) decrypt loop and dispatch them after releasing the lock. The lock is still held continuously across every decrypt in the batch, so no new inter-payload interleaving is introduced; only the non-ratchet dispatch moves out from under it. Matches WA Web, which serializes signal decrypt via a concurrency:1 task scheduler separately from message processing. --- src/message/receive.rs | 112 ++++++++++++++++++++--------------------- 1 file changed, 55 insertions(+), 57 deletions(-) diff --git a/src/message/receive.rs b/src/message/receive.rs index db605eb68..f0af26989 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -609,6 +609,12 @@ impl Client { // Local identity-change detection fires once per batch: the first pkmsg // saves the new key (ReplacedExisting); the rest are NewOrUnchanged. let mut local_identity_reacted = false; + // Successfully decrypted plaintexts, dispatched AFTER the session lock is + // released (see the post-loop dispatch): decode + SKDM/key-share handling + // + durable commit + durability hook + event dispatch don't touch the + // pairwise ratchet the lock guards, so they must not hold it. `enc_type` + // is a `&'static` wire string; the plaintext is owned. + let mut pending_dispatch: Vec<(&'static str, Vec, u8)> = Vec::new(); for payload in payloads { let ciphertext = &payload.ciphertext[..]; @@ -730,35 +736,11 @@ impl Client { local_identity_reacted = true; self.react_to_local_identity_change(sender_encryption_jid); } - let padded_plaintext = decrypted.plaintext; - match self - .clone() - .handle_decrypted_plaintext( - enc_type, - &padded_plaintext, - padding_version, - info, - ) - .await - { - Ok(plaintext_outcome) => { - outcome.decrypted = true; - outcome.dispatched |= plaintext_outcome.dispatched; - outcome.skdm_only |= plaintext_outcome.skdm_only; - } - Err(e) => { - log::warn!( - "[msg:{}] Failed processing plaintext from {}: {e:?}", - info.id, - info.source.sender.observe() - ); - outcome.decrypted = true; - outcome.plaintext_failed = true; - outcome.had_failure = true; - outcome.undecryptable |= - self.handle_plaintext_failure(info, decrypt_fail_mode).await; - } - } + // The ratchet mutation the session lock guards is done; + // defer dispatch (decode, SKDM/key-share, commit, hook, event) + // to the post-loop pass that runs without the lock. + outcome.decrypted = true; + pending_dispatch.push((enc_type, decrypted.plaintext, padding_version)); } Err(e) => { // Handle DuplicatedMessage: This is expected when messages are redelivered during reconnection @@ -865,34 +847,14 @@ impl Client { local_identity_reacted = true; self.react_to_local_identity_change(sender_encryption_jid); } - let padded_plaintext = decrypted.plaintext; - match self - .clone() - .handle_decrypted_plaintext( - enc_type, - &padded_plaintext, - padding_version, - info, - ) - .await - { - Ok(plaintext_outcome) => { - outcome.decrypted = true; - outcome.dispatched |= plaintext_outcome.dispatched; - outcome.skdm_only |= plaintext_outcome.skdm_only; - } - Err(e) => { - log::warn!( - "Failed processing plaintext after identity retry: {e:?}" - ); - outcome.decrypted = true; - outcome.plaintext_failed = true; - outcome.had_failure = true; - outcome.undecryptable |= self - .handle_plaintext_failure(info, decrypt_fail_mode) - .await; - } - } + // Same deferral as the main success path: the + // decrypt is done; dispatch after releasing the lock. + outcome.decrypted = true; + pending_dispatch.push(( + enc_type, + decrypted.plaintext, + padding_version, + )); } Err(retry_err) => { // Handle DuplicatedMessage in retry path: This commonly happens during reconnection @@ -1182,6 +1144,42 @@ impl Client { } } } + + // All pairwise ratchet mutations are done. Release the per-sender session + // lock BEFORE dispatching the decrypted plaintext: dispatch (decode, + // SKDM/key-share handling, durable commit, durability hook, synchronous + // event handlers) does not touch the session ratchet the lock guards — + // PASS 2's group path already runs handle_decrypted_plaintext with no + // session lock. Releasing here means a concurrent same-sender op (a + // retry-receipt session rebuild, a reg-id / base-key session reset) no + // longer waits behind this batch's commit + app-controlled durability hook + // + event dispatch. The batch held the lock continuously across every + // decrypt, so no new inter-payload interleaving is introduced. (The crypto + // timer `_t` drops with the function, spanning dispatch as it did before.) + drop(session_guard.take()); + for (enc_type, padded_plaintext, padding_version) in pending_dispatch { + match self + .clone() + .handle_decrypted_plaintext(enc_type, &padded_plaintext, padding_version, info) + .await + { + Ok(plaintext_outcome) => { + outcome.dispatched |= plaintext_outcome.dispatched; + outcome.skdm_only |= plaintext_outcome.skdm_only; + } + Err(e) => { + log::warn!( + "[msg:{}] Failed processing plaintext from {}: {e:?}", + info.id, + info.source.sender.observe() + ); + outcome.plaintext_failed = true; + outcome.had_failure = true; + outcome.undecryptable |= + self.handle_plaintext_failure(info, decrypt_fail_mode).await; + } + } + } outcome } From 03a2606f223899ecadf597dc5ae89d175350e378 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:36 +0000 Subject: [PATCH 05/30] feat(media): add batched MediaReupload::request_many Recovering N expired-URL media items via request() one at a time meant up to N * 30s of serial waits. request_many registers each item's notification waiter (keyed by its unique message id) and awaits them concurrently behind a bounded window, so a bulk recovery after a long offline period is bounded by ~one timeout instead of the sum. Results preserve input order and one item's failure never aborts the rest. Mirrors WA Web, which runs media work through a concurrency-capped queue rather than a serial loop. request() is unchanged (delegated per item). --- src/features/media_reupload.rs | 38 ++++++++++++++++++++++++++++++++++ 1 file changed, 38 insertions(+) diff --git a/src/features/media_reupload.rs b/src/features/media_reupload.rs index cdf37a2f6..f4c770066 100644 --- a/src/features/media_reupload.rs +++ b/src/features/media_reupload.rs @@ -18,6 +18,14 @@ use wacore_binary::{Jid, JidExt as _}; const MEDIA_RETRY_TIMEOUT: Duration = Duration::from_secs(30); +/// Max media-reupload requests in flight for [`MediaReupload::request_many`]. +/// The per-item work is I/O-light (send a small receipt, park on a notification +/// waiter), so a generous window lets the waits overlap — bulk recovery after a +/// long offline period completes in ~one timeout instead of the sum — while +/// still bounding how many receipts hit the socket/server at once. WA Web caps +/// media work at a similar order (its `ConcurrentPriorityPromiseQueue`). +const MEDIA_REUPLOAD_CONCURRENCY: usize = 32; + /// Error returned by the media reupload request flow. #[derive(Debug, Error)] #[non_exhaustive] @@ -144,6 +152,36 @@ impl<'a> MediaReupload<'a> { req.media_key, )?) } + + /// Request reupload for several messages at once, concurrently. + /// + /// Each request registers its own notification waiter (keyed by the unique + /// message id) and awaits it independently, so a bulk recovery — e.g. many + /// expired-URL media after a long offline period — is bounded by roughly one + /// [`MEDIA_RETRY_TIMEOUT`] instead of the serial sum of per-item waits. + /// Results are returned in the same order as `reqs`; each entry carries that + /// item's success or error (one item failing never aborts the others). + pub async fn request_many( + &self, + reqs: &[MediaReuploadRequest<'_>], + ) -> Vec> { + use futures::StreamExt; + if reqs.is_empty() { + return Vec::new(); + } + + // Stream over owned indices (not a borrow of `reqs` through the + // combinator) and index inside each task, so the fan-out future stays + // Send; collect (index, result) then restore input order. + let mut indexed: Vec<(usize, Result)> = + futures::stream::iter(0..reqs.len()) + .map(|i| async move { (i, self.request(&reqs[i]).await) }) + .buffer_unordered(MEDIA_REUPLOAD_CONCURRENCY) + .collect() + .await; + indexed.sort_by_key(|(i, _)| *i); + indexed.into_iter().map(|(_, res)| res).collect() + } } impl Client { From afdeaf67ff665d1294ed26265e164a2606811b38 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 06/30] perf(connect): don't gate set_passive on the login pre-key upload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The post-login sequence awaited upload_pre_keys_at_login() before set_passive (false), which is what triggers offline message delivery — so on a fresh pairing the user's first messages waited behind key generation + a store + an upload IQ. Uploading one-time pre-keys publishes them for peers' FUTURE sessions; 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. So the upload need not precede going active. Spawn it detached like the signed-pre-key RotateKeyJob right below it (generation-guarded), so offline delivery starts immediately. No-op on reconnect via the persisted server_has_prekeys flag. Matches WA Web, which registers the upload as a passive task (PassiveTaskManager.registerPassiveTask("KeyUpload", ...)) rather than gating active mode on it. --- src/client/node_io.rs | 32 +++++++++++++++++++++++++++----- 1 file changed, 27 insertions(+), 5 deletions(-) diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 412879046..5261ac1ca 100644 --- a/src/client/node_io.rs +++ b/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 From a060647d62924bcc4bbaa7115af5beaad73edc8f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 07/30] perf(status): resolve recipient LIDs concurrently MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit A status post is LID-addressed and its audience can be hundreds of contacts, each resolved to a LID via resolve_recipient_to_lid — a lid_pn cache lookup that falls back to a DB read on a cold cache. These ran one at a time before the post could be assembled. Resolve them behind a bounded fan-out (16), rebuilding the `resolved` vec in the original order since assemble_status_participants is position-sensitive. The network usync / device resolution downstream is unchanged. --- src/send/mod.rs | 28 ++++++++++++++++++++-------- 1 file changed, 20 insertions(+), 8 deletions(-) diff --git a/src/send/mod.rs b/src/send/mod.rs index c3bd29908..100907dbf 100644 --- a/src/send/mod.rs +++ b/src/send/mod.rs @@ -736,18 +736,30 @@ impl Client { } } + use futures::StreamExt; use std::collections::HashMap; - let mut resolved: Vec> = Vec::with_capacity(recipients.len()); + // Resolve each recipient's LID concurrently. A status audience can be + // hundreds of contacts, and resolve_recipient_to_lid falls back to a DB + // read on a cold cache — serializing those was the bottleneck. Each read + // is independent; stream over owned indices and index inside the task so + // `resolved` can be rebuilt in the original order (assemble_status_ + // participants is position-sensitive). Bound (16) matches the other + // resolution fan-outs. + let resolved_indexed: Vec<(usize, Option)> = + futures::stream::iter(0..recipients.len()) + .map(|i| async move { (i, self.resolve_recipient_to_lid(&recipients[i]).await) }) + .buffer_unordered(16) + .collect() + .await; + let mut resolved: Vec> = vec![None; recipients.len()]; let mut lid_to_pn_map: HashMap = HashMap::with_capacity(recipients.len() + 1); - for jid in recipients { - if let Some(lid_jid) = self.resolve_recipient_to_lid(jid).await { - if jid.is_pn() { - lid_to_pn_map.insert(lid_jid.user.clone(), jid.to_non_ad()); + for (i, lid) in resolved_indexed { + if let Some(lid_jid) = lid { + if recipients[i].is_pn() { + lid_to_pn_map.insert(lid_jid.user.clone(), recipients[i].to_non_ad()); } - resolved.push(Some(lid_jid)); - } else { - resolved.push(None); + resolved[i] = Some(lid_jid); } } lid_to_pn_map.insert(own_lid.user.clone(), own_jid.to_non_ad()); From ee95283ed0c1388e63f92860c9fe90c3d800d5ad Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 08/30] docs(sticker_pack): show concurrent zip + thumbnail upload MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The example chained the thumbnail upload behind the zip upload's result, forcing two serial CDN round-trips even though they're independent and share one media key. Show the intended pattern: generate the shared media key up front and tokio::try_join! the two uploads (upload takes &self, so concurrent calls are fine). The library API already supports this — no code change needed, the example just demonstrated the slow path. --- wacore/src/sticker_pack.rs | 17 ++++++++++++----- 1 file changed, 12 insertions(+), 5 deletions(-) diff --git a/wacore/src/sticker_pack.rs b/wacore/src/sticker_pack.rs index 0c309ae0c..bfa6af99a 100644 --- a/wacore/src/sticker_pack.rs +++ b/wacore/src/sticker_pack.rs @@ -9,11 +9,18 @@ //! ]; //! let zip_result = create_sticker_pack_zip("pack-id", &stickers, &cover_webp)?; //! -//! let zip_upload = client.upload(zip_result.zip_bytes.clone(), MediaType::StickerPack, Default::default()).await?; -//! let thumb_upload = client.upload( -//! thumbnail_jpeg, MediaType::StickerPackThumbnail, -//! UploadOptions::new().with_media_key(zip_upload.media_key), -//! ).await?; +//! // The pack zip and its thumbnail are independent uploads that share one +//! // media key. Generate the key up front and upload both concurrently instead +//! // of chaining the thumbnail behind the zip's result — two CDN round-trips +//! // collapse into one wall-clock. (`upload` takes `&self`, so concurrent calls +//! // are fine.) +//! let media_key: [u8; 32] = rand::random(); +//! let (zip_upload, thumb_upload) = futures::try_join!( +//! client.upload(zip_result.zip_bytes.clone(), MediaType::StickerPack, +//! UploadOptions::new().with_media_key(media_key)), +//! client.upload(thumbnail_jpeg, MediaType::StickerPackThumbnail, +//! UploadOptions::new().with_media_key(media_key)), +//! )?; //! //! let metadata = StickerPackMetadata::new(pack_id, "My Pack".into(), "Me".into()); //! let msg = build_sticker_pack_message(&zip_result, &zip_upload.into(), &thumb_upload.into(), metadata)?; From 39ae7a436e91d73d8eed549d2391045b2fd3272c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 09/30] perf(sync): ingest history-sync tasks concurrently, app-state stays serial MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The dedicated sync worker processed every MajorSyncTask one at a time, so a login backlog of independent history-sync chunks ingested serially — each chunk's download + gzip inflate + protobuf scan + secret/tctoken store blocking the next. History-sync tasks are independent (order-free upserts; the dispatched event carries chunk_order for consumers), so process them concurrently behind a small semaphore. App-state tasks stay strictly serial and inline — same-collection patch application is order-sensitive. The permit is acquired in the recv loop so a burst applies backpressure rather than spawning unbounded tasks that each pin a compressed blob. Cap is deliberately low (2): each task transiently holds a decompressed history blob and the connect path is peak-memory-conscious. WA Web caps history-sync chunks similarly (histSyncChunk=3); the in-flight counter that gates startup-sync completion is an atomic, so concurrent finishes are safe. --- src/bot.rs | 27 ++++++++++++++++++++++++++- 1 file changed, 26 insertions(+), 1 deletion(-) diff --git a/src/bot.rs b/src/bot.rs index 3928960d0..4a39048ae 100644 --- a/src/bot.rs +++ b/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."); })) From bf2f51af30d40e1b45cb160551b054224f3dc249 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 10/30] perf(session): probe has_session concurrently in ensure_sessions MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ensure_sessions_inner checked each device's session one await at a time before the (already batched) prekey fetch. On a cold-cache multi-recipient ensure — status / group session setup over many devices — that serialized the per-device DB reads. Fan the probes out behind a bounded window (16). Warm hits still serialize on the signal cache mutex, so this only helps the cold-cache DB-miss portion, but the probes are independent and their order is irrelevant (misses are chunked for the fetch). --- src/client/sessions.rs | 41 +++++++++++++++++++++++++++-------------- 1 file changed, 27 insertions(+), 14 deletions(-) diff --git a/src/client/sessions.rs b/src/client/sessions.rs index 0798f7fa3..5a5412dfb 100644 --- a/src/client/sessions.rs +++ b/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 = 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(()); From e2267b680a07a013431e436f22b27a4f9069fe45 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 11/30] perf(prekeys): load companion account identities concurrently collect_account_identities runs before every prekey fetch and loaded each companion's device-0 identity (signal-cache / DB read) one at a time. On a cold group send the keyless-companion set can be large, so fan the loads out behind a bounded window instead of serializing them ahead of the already-batched fetch IQ. --- src/prekeys.rs | 24 +++++++++++++++++------- 1 file changed, 17 insertions(+), 7 deletions(-) diff --git a/src/prekeys.rs b/src/prekeys.rs index 9b58fe3cd..a73398e0c 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -177,13 +177,23 @@ impl Client { &self, jids: &[Jid], ) -> std::collections::HashMap { - let mut map = std::collections::HashMap::new(); - for jid in jids.iter().filter(|j| j.device != 0) { - if let Some(id) = self.load_account_identity(jid).await { - map.insert(jid.normalize_for_prekey_bundle(), id); - } - } - map + use futures::StreamExt; + // Materialize the companion (device != 0) subset so the stream owns its + // items (borrowing `jids` through buffer_unordered trips the future's Send + // bound). Each load is an independent signal-cache/DB read; on a cold group + // send the keyless-companion set can be large, so fan them out instead of + // serializing before the (already batched) prekey fetch IQ. + let companions: Vec = jids.iter().filter(|j| j.device != 0).cloned().collect(); + futures::stream::iter(companions) + .map(|jid| async move { + self.load_account_identity(&jid) + .await + .map(|id| (jid.normalize_for_prekey_bundle(), id)) + }) + .buffer_unordered(16) + .filter_map(|entry| async move { entry }) + .collect() + .await } /// Load a companion's account (device 0) identity from the store, for use as From 7584782a068eccb847be4e4f7228d169478807c4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 12/30] perf(contacts): run is_on_whatsapp PN and LID queries concurrently A mixed PN+LID input issued the two existence-check IQs sequentially. They're independent queries, so run them concurrently with tokio::join! when both are present (each still short-circuits to empty when its user list is). --- src/features/contacts.rs | 46 ++++++++++++++++++++++++++++------------ 1 file changed, 33 insertions(+), 13 deletions(-) diff --git a/src/features/contacts.rs b/src/features/contacts.rs index 8b890c2e3..3c13e0e95 100644 --- a/src/features/contacts.rs +++ b/src/features/contacts.rs @@ -145,19 +145,39 @@ impl<'a> Contacts<'a> { } } - let mut results = Vec::new(); - - if !pn_users.is_empty() { - let sid = self.client.generate_request_id(); - let spec = IsOnWhatsAppSpec::new(pn_users, sid, IsOnWhatsAppQueryType::Pn); - results.extend(self.client.execute(spec).await?); - } - - if !lid_users.is_empty() { - let sid = self.client.generate_request_id(); - let spec = IsOnWhatsAppSpec::new(lid_users, sid, IsOnWhatsAppQueryType::Lid); - results.extend(self.client.execute(spec).await?); - } + // PN and LID existence use different protocols (two independent IQs), so + // when a mixed input produces both, run them concurrently. + let pn_fut = async { + if pn_users.is_empty() { + Ok(Vec::new()) + } else { + let sid = self.client.generate_request_id(); + self.client + .execute(IsOnWhatsAppSpec::new( + pn_users, + sid, + IsOnWhatsAppQueryType::Pn, + )) + .await + } + }; + let lid_fut = async { + if lid_users.is_empty() { + Ok(Vec::new()) + } else { + let sid = self.client.generate_request_id(); + self.client + .execute(IsOnWhatsAppSpec::new( + lid_users, + sid, + IsOnWhatsAppQueryType::Lid, + )) + .await + } + }; + let (pn_results, lid_results) = futures::join!(pn_fut, lid_fut); + let mut results = pn_results?; + results.extend(lid_results?); self.persist_lid_mappings(results.iter().map(forward_lid_pair)) .await; From 368795fef7fde3dac90d3760d79b2ca3ae00913f Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:21:37 +0000 Subject: [PATCH 13/30] perf(retry): resolve chat and sender JIDs concurrently for the cache key make_retry_cache_key resolved the chat and sender JIDs to their encryption namespace one after the other; they're independent LID/PN lookups, so join them. --- src/message/retry.rs | 7 +++++-- 1 file changed, 5 insertions(+), 2 deletions(-) diff --git a/src/message/retry.rs b/src/message/retry.rs index a42e20a88..a69ed4f9e 100644 --- a/src/message/retry.rs +++ b/src/message/retry.rs @@ -154,8 +154,11 @@ impl Client { msg_id: &str, sender: &Jid, ) -> String { - let chat = self.resolve_encryption_jid(chat).await; - let sender = self.resolve_encryption_jid(sender).await; + // Two independent LID/PN resolves for different JIDs — run concurrently. + let (chat, sender) = futures::join!( + self.resolve_encryption_jid(chat), + self.resolve_encryption_jid(sender), + ); // +40 covers @server suffixes, :device, separators for two JIDs let mut key = String::with_capacity(chat.user.len() + msg_id.len() + sender.user.len() + 40); From ee4aeb357c92d7968eb3be8f65ea21c693e1c6ea Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:51:35 +0000 Subject: [PATCH 14/30] fix(media): give each download() attempt a fresh buffer MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Routing download() through the shared-writer path meant retries/host-failover reused one Cursor>, which seeks to 0 but does not truncate: a failed host that wrote a longer body (e.g. a CDN error page decrypting to more bytes before its MAC fails) would leave a stale tail behind a shorter successful retry, and into_inner() would return it as valid data — a silent corruption the old fresh-Vec-per-request path never had. Restore fresh-buffer-per-attempt via download_media_with_retry while keeping the streaming win: each attempt streams (CDN read + decrypt interleaved) into its own Cursor. Also document the empty-writer contract on download_to_writer, whose File path has the same no-truncate-on-seek behavior. --- src/download.rs | 109 +++++++++++++++++++++++++++++++++++++++++------- 1 file changed, 94 insertions(+), 15 deletions(-) diff --git a/src/download.rs b/src/download.rs index 82b38cc16..de89bc21b 100644 --- a/src/download.rs +++ b/src/download.rs @@ -123,6 +123,74 @@ impl DownloadRequestError { } } +/// Auth-refresh + host-failover retry loop that returns the decrypted bytes. +/// Unlike [`download_to_writer_with_retry`] each attempt gets a FRESH buffer +/// (the executor allocates its own), so a failed host that wrote a longer body +/// (e.g. a CDN error page that decrypts to more bytes before its MAC fails) +/// can't leave a stale tail behind a shorter successful retry. +async fn download_media_with_retry< + PrepareRequests, + PrepareRequestsFut, + InvalidateMediaConn, + InvalidateMediaConnFut, + ExecuteRequest, + ExecuteRequestFut, +>( + mut prepare_requests: PrepareRequests, + mut invalidate_media_conn: InvalidateMediaConn, + mut execute_request: ExecuteRequest, +) -> Result> +where + PrepareRequests: FnMut(bool) -> PrepareRequestsFut, + PrepareRequestsFut: + std::future::Future>>, + InvalidateMediaConn: FnMut() -> InvalidateMediaConnFut, + InvalidateMediaConnFut: std::future::Future, + ExecuteRequest: FnMut(wacore::download::DownloadRequest) -> ExecuteRequestFut, + ExecuteRequestFut: + std::future::Future, DownloadRequestError>>, +{ + let mut force_refresh = false; + let mut last_err: Option = None; + + for attempt in 0..=MEDIA_AUTH_REFRESH_RETRY_ATTEMPTS { + let requests = prepare_requests(force_refresh).await?; + let mut retry_with_fresh_auth = false; + + for request in requests { + match execute_request(request.clone()).await { + Ok(data) => return Ok(data), + Err(err) if (err.is_auth() || err.is_not_found()) && attempt == 0 => { + // Auth error or 404/410 (expired URL): refresh media conn and re-derive URLs. + invalidate_media_conn().await; + force_refresh = true; + retry_with_fresh_auth = true; + break; + } + Err(err) if err.is_auth() || err.is_not_found() => return Err(err.into_anyhow()), + Err(err) => { + let err = err.into_anyhow(); + log::warn!( + "Failed to download from URL {}: {:?}. Trying next host.", + request.url, + err + ); + last_err = Some(err); + } + } + } + + if !retry_with_fresh_auth { + break; + } + } + + match last_err { + Some(err) => Err(err), + None => Err(anyhow!("Failed to download from all available media hosts")), + } +} + async fn download_to_writer_with_retry< W, PrepareRequests, @@ -202,25 +270,30 @@ impl Client { tracing::instrument(name = "wa.media.download", level = "debug", skip_all, err(Debug)) )] pub async fn download(&self, downloadable: &dyn Downloadable) -> Result> { - // Route through the streaming writer path (into an in-memory buffer) - // instead of "fetch the whole ciphertext, then decrypt it in a second - // pass". The streaming path interleaves the CDN read with the AES/HMAC - // decrypt in a single blocking pass, overlapping network with CPU and - // roughly halving peak memory (no full-ciphertext buffer resident - // alongside the plaintext). `download_to_writer` falls back to a buffered - // fetch+decrypt automatically for non-streaming HTTP clients, so this is - // strictly >= the previous behavior. Pre-size the buffer to the plaintext - // length (the decrypted output size) when known, capped so a bogus - // server-declared length can't drive a huge speculative allocation. + // Stream each attempt into a FRESH in-memory buffer: the CDN read and the + // AES/HMAC decrypt interleave in one blocking pass (overlapping network + // with CPU, ~halving peak memory vs. fetch-whole-then-decrypt), and a new + // buffer per attempt means a failed host that wrote a longer body can't + // leave a stale tail behind a shorter successful retry. Pre-size to the + // declared plaintext length, capped so a bogus length can't drive a huge + // speculative allocation. let cap = downloadable .file_length() .unwrap_or(0) .min(DOWNLOAD_PREALLOC_CAP) as usize; - let writer = std::io::Cursor::new(Vec::with_capacity(cap)); - Ok(self - .download_to_writer(downloadable, writer) - .await? - .into_inner()) + download_media_with_retry( + |force| self.prepare_requests(downloadable, force), + || async { self.invalidate_media_conn().await }, + |request| async move { + let writer = std::io::Cursor::new(Vec::with_capacity(cap)); + match self.streaming_download_and_decrypt(&request, writer).await { + Ok((writer, Ok(()))) => Ok(writer.into_inner()), + Ok((_, Err(e))) => Err(e), + Err(e) => Err(DownloadRequestError::other(e)), + } + }, + ) + .await } /// Fetch a first-party sticker pack's metadata and sticker list from the CDN. @@ -278,6 +351,12 @@ impl Client { /// The entire HTTP download, decryption, and file write happen in a single /// blocking thread. The writer is seeked back to position 0 before returning. /// + /// The `writer` MUST start empty. Retries/host-failover seek back to 0 and + /// rewrite but do NOT truncate, so a writer that already held more bytes than + /// the decrypted payload would keep a stale tail past the valid data. (The + /// in-memory [`Self::download`] gives every attempt a fresh buffer for exactly + /// this reason.) + /// /// Memory usage: ~40KB regardless of file size (8KB read buffer + decrypt state). #[cfg_attr( feature = "tracing", From b4fd1c89d1bb8651533771b7c6906b22100dc500 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:51:35 +0000 Subject: [PATCH 15/30] fix(recv): defer dispatch only for single-payload session batches MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The lock-narrowing deferred regular decrypt successes to a post-loop dispatch, but the PN->LID migration fallback still dispatches inline mid-loop. In a multi-payload batch that could deliver a later migrated payload before an earlier deferred one, inverting intra-stanza order. Defer only when the batch has a single payload (the overwhelming common case) — then there is nothing to reorder, and the lock-release-before-dispatch win still applies. Multi-payload batches dispatch inline under the lock as before, preserving order. Shared dispatch logic extracted into dispatch_session_plaintext. --- src/message/receive.rs | 122 ++++++++++++++++++++++++++++------------- 1 file changed, 85 insertions(+), 37 deletions(-) diff --git a/src/message/receive.rs b/src/message/receive.rs index f0af26989..0f3a70310 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -575,6 +575,41 @@ impl Client { } #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.session_decrypt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %sender_encryption_jid.observe(), msg_id = %info.id)))] + /// Dispatch one decrypted session plaintext (decode, SKDM/key-share, commit, + /// hook, event) and fold the result into `outcome`. None of this touches the + /// pairwise ratchet, so it runs without the per-sender session lock. + async fn dispatch_session_plaintext( + self: &Arc, + enc_type: &str, + padded_plaintext: &[u8], + padding_version: u8, + info: &Arc, + decrypt_fail_mode: crate::types::events::DecryptFailMode, + outcome: &mut SessionBatchOutcome, + ) { + match self + .clone() + .handle_decrypted_plaintext(enc_type, padded_plaintext, padding_version, info) + .await + { + Ok(plaintext_outcome) => { + outcome.dispatched |= plaintext_outcome.dispatched; + outcome.skdm_only |= plaintext_outcome.skdm_only; + } + Err(e) => { + log::warn!( + "[msg:{}] Failed processing plaintext from {}: {e:?}", + info.id, + info.source.sender.observe() + ); + outcome.plaintext_failed = true; + outcome.had_failure = true; + outcome.undecryptable |= + self.handle_plaintext_failure(info, decrypt_fail_mode).await; + } + } + } + pub(crate) async fn process_session_enc_batch( self: Arc, payloads: &[EncPayload], @@ -609,11 +644,15 @@ impl Client { // Local identity-change detection fires once per batch: the first pkmsg // saves the new key (ReplacedExisting); the rest are NewOrUnchanged. let mut local_identity_reacted = false; - // Successfully decrypted plaintexts, dispatched AFTER the session lock is - // released (see the post-loop dispatch): decode + SKDM/key-share handling - // + durable commit + durability hook + event dispatch don't touch the - // pairwise ratchet the lock guards, so they must not hold it. `enc_type` - // is a `&'static` wire string; the plaintext is owned. + // For a single-payload batch (the overwhelming common case) the decrypted + // plaintext is deferred here and dispatched AFTER the session lock is + // released — decode / SKDM / commit / hook / event don't touch the ratchet + // the lock guards, so releasing early cuts contention with a concurrent + // same-sender op (a retry-receipt session rebuild). Multi-payload batches + // dispatch inline under the lock instead, preserving intra-stanza order: + // the PN→LID migration fallback dispatches inline mid-loop, so deferring + // only some payloads could otherwise deliver a later one before an earlier. + let defer_dispatch = payloads.len() == 1; let mut pending_dispatch: Vec<(&'static str, Vec, u8)> = Vec::new(); for payload in payloads { @@ -736,11 +775,21 @@ impl Client { local_identity_reacted = true; self.react_to_local_identity_change(sender_encryption_jid); } - // The ratchet mutation the session lock guards is done; - // defer dispatch (decode, SKDM/key-share, commit, hook, event) - // to the post-loop pass that runs without the lock. + // The ratchet mutation the session lock guards is done. outcome.decrypted = true; - pending_dispatch.push((enc_type, decrypted.plaintext, padding_version)); + if defer_dispatch { + pending_dispatch.push((enc_type, decrypted.plaintext, padding_version)); + } else { + self.dispatch_session_plaintext( + enc_type, + &decrypted.plaintext, + padding_version, + info, + decrypt_fail_mode, + &mut outcome, + ) + .await; + } } Err(e) => { // Handle DuplicatedMessage: This is expected when messages are redelivered during reconnection @@ -847,14 +896,25 @@ impl Client { local_identity_reacted = true; self.react_to_local_identity_change(sender_encryption_jid); } - // Same deferral as the main success path: the - // decrypt is done; dispatch after releasing the lock. + // Same deferral as the main success path. outcome.decrypted = true; - pending_dispatch.push(( - enc_type, - decrypted.plaintext, - padding_version, - )); + if defer_dispatch { + pending_dispatch.push(( + enc_type, + decrypted.plaintext, + padding_version, + )); + } else { + self.dispatch_session_plaintext( + enc_type, + &decrypted.plaintext, + padding_version, + info, + decrypt_fail_mode, + &mut outcome, + ) + .await; + } } Err(retry_err) => { // Handle DuplicatedMessage in retry path: This commonly happens during reconnection @@ -1158,27 +1218,15 @@ impl Client { // timer `_t` drops with the function, spanning dispatch as it did before.) drop(session_guard.take()); for (enc_type, padded_plaintext, padding_version) in pending_dispatch { - match self - .clone() - .handle_decrypted_plaintext(enc_type, &padded_plaintext, padding_version, info) - .await - { - Ok(plaintext_outcome) => { - outcome.dispatched |= plaintext_outcome.dispatched; - outcome.skdm_only |= plaintext_outcome.skdm_only; - } - Err(e) => { - log::warn!( - "[msg:{}] Failed processing plaintext from {}: {e:?}", - info.id, - info.source.sender.observe() - ); - outcome.plaintext_failed = true; - outcome.had_failure = true; - outcome.undecryptable |= - self.handle_plaintext_failure(info, decrypt_fail_mode).await; - } - } + self.dispatch_session_plaintext( + enc_type, + &padded_plaintext, + padding_version, + info, + decrypt_fail_mode, + &mut outcome, + ) + .await; } outcome } From ec08405f1f608fdb26d5234d525af66c5ad6aee4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:51:35 +0000 Subject: [PATCH 16/30] perf(contacts): fail fast when an is_on_whatsapp query errors MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit futures::join! polls both the PN and LID queries to completion even if one errors, so a failing query would still wait on the other's full round-trip — a latency regression vs. the old sequential fail-on-first. Use try_join! to return on the first error and drop the other in-flight future. --- src/features/contacts.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/features/contacts.rs b/src/features/contacts.rs index 3c13e0e95..e085c29cf 100644 --- a/src/features/contacts.rs +++ b/src/features/contacts.rs @@ -175,9 +175,11 @@ impl<'a> Contacts<'a> { .await } }; - let (pn_results, lid_results) = futures::join!(pn_fut, lid_fut); - let mut results = pn_results?; - results.extend(lid_results?); + // try_join! fails fast: an error from either query returns immediately and + // drops the other in-flight future (the sequential version failed on the + // first error too, so join! would have been a latency regression there). + let (mut results, lid_results) = futures::try_join!(pn_fut, lid_fut)?; + results.extend(lid_results); self.persist_lid_mappings(results.iter().map(forward_lid_pair)) .await; From d5c1acedac39edff41008206e9c3c030846a24b8 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:51:35 +0000 Subject: [PATCH 17/30] perf(appstate): dedup external blob downloads by directPath The result map is keyed by directPath, so two patches referencing the same blob would each trigger an independent CDN GET racing to write the same key, wasting a parallel slot. Skip already-seen directPaths when building the fetch list. --- src/client/app_state.rs | 46 +++++++++++++++++------------------------ 1 file changed, 19 insertions(+), 27 deletions(-) diff --git a/src/client/app_state.rs b/src/client/app_state.rs index cc6daf698..919091513 100644 --- a/src/client/app_state.rs +++ b/src/client/app_state.rs @@ -2,15 +2,10 @@ 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)`. +/// 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 { @@ -28,43 +23,40 @@ 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`). + /// 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> { use futures::StreamExt; - // Blob context, kept only so a failed download logs the same message the - // serial version did (snapshot name vs. patch version). + // Kept only so a failed download logs the right message (snapshot vs patch). 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). + // 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 - && ext.direct_path.is_some() + && 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() - && ext.direct_path.is_some() + && let Some(path) = ext.direct_path.as_deref() + && seen_paths.insert(path) { let v = patch .version From edc177c34d5b4112eada3d29c68ab29e4a7b8fe1 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 05:51:35 +0000 Subject: [PATCH 18/30] style: clarify sync intake log and trim verbose comments MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The "Sync worker shutting down" log fired when the intake loop ended, but detached history-sync tasks may still run — reworded to say so. Also condense the paragraph-length comments added in this branch to be why-focused per AGENTS.md ("don't be so verbose, only explain why, not what"). --- src/bot.rs | 20 +++++++++----------- src/client/node_io.rs | 13 +++++-------- src/client/sessions.rs | 8 +++----- src/prekeys.rs | 8 +++----- src/send/mod.rs | 10 +++------- src/usync.rs | 23 +++++++---------------- 6 files changed, 30 insertions(+), 52 deletions(-) diff --git a/src/bot.rs b/src/bot.rs index 4a39048ae..48dd416bb 100644 --- a/src/bot.rs +++ b/src/bot.rs @@ -467,16 +467,12 @@ 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. + // 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)); @@ -503,7 +499,9 @@ impl Bot { 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(); } diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 5261ac1ca..1b954f2f1 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -743,14 +743,11 @@ impl Client { debug!("Skipping passive tasks: connection closed"); return; } - // 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. + // 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; diff --git a/src/client/sessions.rs b/src/client/sessions.rs index 5a5412dfb..a080d11b2 100644 --- a/src/client/sessions.rs +++ b/src/client/sessions.rs @@ -289,11 +289,9 @@ impl Client { let device_snapshot = self.persistence_manager.get_device_snapshot(); - // 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). + // 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 = futures::stream::iter(jids) diff --git a/src/prekeys.rs b/src/prekeys.rs index a73398e0c..b4b3f65dc 100644 --- a/src/prekeys.rs +++ b/src/prekeys.rs @@ -178,11 +178,9 @@ impl Client { jids: &[Jid], ) -> std::collections::HashMap { use futures::StreamExt; - // Materialize the companion (device != 0) subset so the stream owns its - // items (borrowing `jids` through buffer_unordered trips the future's Send - // bound). Each load is an independent signal-cache/DB read; on a cold group - // send the keyless-companion set can be large, so fan them out instead of - // serializing before the (already batched) prekey fetch IQ. + // Fan out the per-companion identity loads (independent cache/DB reads) — + // the keyless-companion set can be large on a cold group send. Owned Vec so + // the stream doesn't borrow `jids` through buffer_unordered (Send bound). let companions: Vec = jids.iter().filter(|j| j.device != 0).cloned().collect(); futures::stream::iter(companions) .map(|jid| async move { diff --git a/src/send/mod.rs b/src/send/mod.rs index 100907dbf..1b155b2b9 100644 --- a/src/send/mod.rs +++ b/src/send/mod.rs @@ -738,13 +738,9 @@ impl Client { use futures::StreamExt; use std::collections::HashMap; - // Resolve each recipient's LID concurrently. A status audience can be - // hundreds of contacts, and resolve_recipient_to_lid falls back to a DB - // read on a cold cache — serializing those was the bottleneck. Each read - // is independent; stream over owned indices and index inside the task so - // `resolved` can be rebuilt in the original order (assemble_status_ - // participants is position-sensitive). Bound (16) matches the other - // resolution fan-outs. + // Resolve recipient LIDs concurrently (a status audience can be hundreds of + // contacts, each a cold-cache DB read). Stream over indices and rebuild + // `resolved` in order — assemble_status_participants is position-sensitive. let resolved_indexed: Vec<(usize, Option)> = futures::stream::iter(0..recipients.len()) .map(|i| async move { (i, self.resolve_recipient_to_lid(&recipients[i]).await) }) diff --git a/src/usync.rs b/src/usync.rs index 1a2a4bd16..8fcf3d28a 100644 --- a/src/usync.rs +++ b/src/usync.rs @@ -14,23 +14,14 @@ impl Client { let mut jids_to_fetch: HashSet = HashSet::with_capacity(jids.len()); let mut all_devices = Vec::with_capacity(jids.len() * 2); - // Resolve each user's device list from the registry concurrently. The - // network usync is already batched into one IQ below; this is the LOCAL - // read scan (in-memory cache, DB fallback per user) that a large group - // would otherwise serialize over 256+ members on a cold cache. Each read - // takes `&self` and is independent, and the resulting device set order is - // irrelevant (the phash sorts, and the encrypt fan-out is order-agnostic), - // so a bounded fan-out is safe. Bound (16) matches the send encrypt - // fan-out and `groups::fill_participant_pns`. - // - // get_devices_from_registry returns None for an empty record (never a - // valid set — WA Web always keeps device 0), so a corrupted empty row - // falls through to the network below instead of being trusted. + // Resolve the LOCAL registry scan concurrently (the network usync below is + // already one batched IQ) — a cold-cache large group would otherwise + // serialize 256+ per-user cache/DB reads. Order is irrelevant (phash sorts, + // encrypt fan-out is order-agnostic). A None result means an empty/corrupt + // record, which falls through to the network below (WA Web always keeps + // device 0). Materialize to an owned Vec first so the stream doesn't borrow + // `jids` through buffer_unordered (Send bound). use futures::StreamExt; - // Materialize the non-AD users into an owned Vec so the stream owns its - // items (borrowing `jids` through `buffer_unordered` trips the future's - // higher-ranked Send bound — same reason `groups::fill_participant_pns` - // collects `pending` first). let non_ad: Vec = jids.iter().map(|j| j.to_non_ad()).collect(); let resolved: Vec<(Jid, Option>)> = futures::stream::iter(non_ad) .map(|jid| async move { From 3a15af803e6ff5ed707471c18eb23f5574aa21a2 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 06:03:14 +0000 Subject: [PATCH 19/30] fix(recv): keep tracing::instrument on process_session_enc_batch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Inserting dispatch_session_plaintext landed between the #[cfg_attr(tracing, instrument(... %sender_encryption_jid ...))] attribute and process_session_enc_batch, so the attribute decorated the helper (which has no sender_encryption_jid) — a compile error under --features tracing, and the session-decrypt span was silently dropped. Move the attribute back onto process_session_enc_batch. --- src/message/receive.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/message/receive.rs b/src/message/receive.rs index 0f3a70310..2dce26129 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -574,7 +574,6 @@ impl Client { } } - #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.session_decrypt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %sender_encryption_jid.observe(), msg_id = %info.id)))] /// Dispatch one decrypted session plaintext (decode, SKDM/key-share, commit, /// hook, event) and fold the result into `outcome`. None of this touches the /// pairwise ratchet, so it runs without the per-sender session lock. @@ -610,6 +609,7 @@ impl Client { } } + #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.session_decrypt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %sender_encryption_jid.observe(), msg_id = %info.id)))] pub(crate) async fn process_session_enc_batch( self: Arc, payloads: &[EncPayload], From 95b9324732e1186ad738a64135a55754b43c2091 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 06:03:14 +0000 Subject: [PATCH 20/30] fix(contacts): use join! not try_join! for is_on_whatsapp MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit try_join! drops the sibling IQ future the instant one query errors, but send_and_wait_iq removes its response_waiters entry only on send-failure/timeout/shutdown — not on cancellation-via-drop — so the abandoned waiter lingers and suppresses keepalives. Await both with join! so each cleans up its own waiter; the concurrency win (one round-trip on mixed PN+LID input) is unchanged and the extra wait only affects the rare error path. --- src/features/contacts.rs | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/src/features/contacts.rs b/src/features/contacts.rs index e085c29cf..6a715ea62 100644 --- a/src/features/contacts.rs +++ b/src/features/contacts.rs @@ -175,11 +175,14 @@ impl<'a> Contacts<'a> { .await } }; - // try_join! fails fast: an error from either query returns immediately and - // drops the other in-flight future (the sequential version failed on the - // first error too, so join! would have been a latency regression there). - let (mut results, lid_results) = futures::try_join!(pn_fut, lid_fut)?; - results.extend(lid_results); + // join!, NOT try_join!: a fail-fast try_join! would drop the sibling IQ + // future the instant one errored, leaking its `response_waiters` entry — + // send_and_wait_iq only removes the waiter on send-failure/timeout/shutdown, + // not on cancellation-via-drop, and a lingering waiter suppresses + // keepalives. Awaiting both lets each clean up its own waiter. + let (pn_results, lid_results) = futures::join!(pn_fut, lid_fut); + let mut results = pn_results?; + results.extend(lid_results?); self.persist_lid_mappings(results.iter().map(forward_lid_pair)) .await; From 68be4c2be2a64e52c0dc73d1923c8946d47e6820 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 06:04:35 +0000 Subject: [PATCH 21/30] test(media): cover download_media_with_retry auth-refresh retry download() routes through download_media_with_retry (fresh buffer per attempt), a separate engine from download_to_writer_with_retry, so restore the direct auth-refresh + host-failover test for it (the prior comment wrongly attributed that coverage to the writer path's test). --- src/download.rs | 77 ++++++++++++++++++++++++++++++++++++++++++++++--- 1 file changed, 73 insertions(+), 4 deletions(-) diff --git a/src/download.rs b/src/download.rs index de89bc21b..4c8d88d5f 100644 --- a/src/download.rs +++ b/src/download.rs @@ -634,10 +634,79 @@ mod tests { ); } - // The buffered `download()` now routes through `download_to_writer`, so its - // auth-refresh + host-failover retry behavior is exercised by - // `download_to_writer_retries_with_forced_media_conn_refresh_after_auth_error` - // below (same `download_to_writer_with_retry` engine). + // `download()` uses `download_media_with_retry` (fresh buffer per attempt); + // cover its auth-refresh + host-failover retry behavior directly. + #[tokio::test] + async fn download_retries_with_forced_media_conn_refresh_after_auth_error() { + let body = b"download me".to_vec(); + let downloadable = PlaintextDownloadable { + direct_path: "/v/t62.7118-24/123".to_string(), + file_sha256: plaintext_sha256(&body), + }; + let first_conn = media_conn("stale-auth", &["cdn1.example.com"]); + let refreshed_conn = media_conn("fresh-auth", &["cdn2.example.com"]); + let refresh_calls = Arc::new(Mutex::new(Vec::new())); + let invalidations = Arc::new(Mutex::new(0usize)); + let seen_urls = Arc::new(Mutex::new(Vec::new())); + + let downloaded = download_media_with_retry( + { + let refresh_calls = Arc::clone(&refresh_calls); + let downloadable = &downloadable; + move |force| { + let refresh_calls = Arc::clone(&refresh_calls); + let first_conn = first_conn.clone(); + let refreshed_conn = refreshed_conn.clone(); + async move { + refresh_calls.lock().await.push(force); + let media_conn = if force { refreshed_conn } else { first_conn }; + DownloadUtils::prepare_download_requests( + downloadable, + &wacore::download::MediaConnection::from(&media_conn), + ) + } + } + }, + { + let invalidations = Arc::clone(&invalidations); + move || { + let invalidations = Arc::clone(&invalidations); + async move { + *invalidations.lock().await += 1; + } + } + }, + { + let seen_urls = Arc::clone(&seen_urls); + let body = body.clone(); + move |request| { + let seen_urls = Arc::clone(&seen_urls); + let body = body.clone(); + let url = request.url.clone(); + async move { + seen_urls.lock().await.push(url.clone()); + if url.contains("stale-auth") { + Err(DownloadRequestError::auth(401)) + } else { + Ok(body) + } + } + } + }, + ) + .await + .expect("download should succeed after refreshing media auth"); + + assert_eq!(downloaded, body); + assert_eq!(*refresh_calls.lock().await, vec![false, true]); + assert_eq!(*invalidations.lock().await, 1); + + let seen_urls = seen_urls.lock().await.clone(); + assert_eq!(seen_urls.len(), 2); + assert!(seen_urls[0].contains("auth=stale-auth")); + assert!(seen_urls[1].contains("auth=fresh-auth")); + } + #[tokio::test] async fn download_to_writer_retries_with_forced_media_conn_refresh_after_auth_error() { let body = b"stream me".to_vec(); From a9f72e06bf1f585faeef27dfbe0061e768647d26 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 06:11:55 +0000 Subject: [PATCH 22/30] revert(recv): keep the per-sender session lock through dispatch MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Reverts the session-lock narrowing in process_session_enc_batch. The pairwise Signal session ratchet is shared across chats for one sender, so releasing the lock after decrypt let a same-sender message in another (concurrent) chat lane advance the same ratchet and flush the signal cache — persisting this message's ratchet advance — before this message's durable inbound row was committed. A crash in that window makes the redelivered message look like a duplicate with no buffered row, so it is acked instead of replayed, violating the inbound durability hook's at-least-once contract. The commit-batch durability model was designed around holding this lock through handle_decrypted_plaintext. The contention win (not blocking a concurrent same-sender retry-receipt rebuild) did not justify that risk on the most correctness-critical path. Restores the original dispatch-under-lock behavior verbatim. --- src/message/receive.rs | 140 ++++++++++++++--------------------------- 1 file changed, 47 insertions(+), 93 deletions(-) diff --git a/src/message/receive.rs b/src/message/receive.rs index 2dce26129..db605eb68 100644 --- a/src/message/receive.rs +++ b/src/message/receive.rs @@ -574,41 +574,6 @@ impl Client { } } - /// Dispatch one decrypted session plaintext (decode, SKDM/key-share, commit, - /// hook, event) and fold the result into `outcome`. None of this touches the - /// pairwise ratchet, so it runs without the per-sender session lock. - async fn dispatch_session_plaintext( - self: &Arc, - enc_type: &str, - padded_plaintext: &[u8], - padding_version: u8, - info: &Arc, - decrypt_fail_mode: crate::types::events::DecryptFailMode, - outcome: &mut SessionBatchOutcome, - ) { - match self - .clone() - .handle_decrypted_plaintext(enc_type, padded_plaintext, padding_version, info) - .await - { - Ok(plaintext_outcome) => { - outcome.dispatched |= plaintext_outcome.dispatched; - outcome.skdm_only |= plaintext_outcome.skdm_only; - } - Err(e) => { - log::warn!( - "[msg:{}] Failed processing plaintext from {}: {e:?}", - info.id, - info.source.sender.observe() - ); - outcome.plaintext_failed = true; - outcome.had_failure = true; - outcome.undecryptable |= - self.handle_plaintext_failure(info, decrypt_fail_mode).await; - } - } - } - #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.recv.session_decrypt", level = "debug", skip_all, fields(chat = %info.source.chat.observe(), sender = %sender_encryption_jid.observe(), msg_id = %info.id)))] pub(crate) async fn process_session_enc_batch( self: Arc, @@ -644,16 +609,6 @@ impl Client { // Local identity-change detection fires once per batch: the first pkmsg // saves the new key (ReplacedExisting); the rest are NewOrUnchanged. let mut local_identity_reacted = false; - // For a single-payload batch (the overwhelming common case) the decrypted - // plaintext is deferred here and dispatched AFTER the session lock is - // released — decode / SKDM / commit / hook / event don't touch the ratchet - // the lock guards, so releasing early cuts contention with a concurrent - // same-sender op (a retry-receipt session rebuild). Multi-payload batches - // dispatch inline under the lock instead, preserving intra-stanza order: - // the PN→LID migration fallback dispatches inline mid-loop, so deferring - // only some payloads could otherwise deliver a later one before an earlier. - let defer_dispatch = payloads.len() == 1; - let mut pending_dispatch: Vec<(&'static str, Vec, u8)> = Vec::new(); for payload in payloads { let ciphertext = &payload.ciphertext[..]; @@ -775,20 +730,34 @@ impl Client { local_identity_reacted = true; self.react_to_local_identity_change(sender_encryption_jid); } - // The ratchet mutation the session lock guards is done. - outcome.decrypted = true; - if defer_dispatch { - pending_dispatch.push((enc_type, decrypted.plaintext, padding_version)); - } else { - self.dispatch_session_plaintext( + let padded_plaintext = decrypted.plaintext; + match self + .clone() + .handle_decrypted_plaintext( enc_type, - &decrypted.plaintext, + &padded_plaintext, padding_version, info, - decrypt_fail_mode, - &mut outcome, ) - .await; + .await + { + Ok(plaintext_outcome) => { + outcome.decrypted = true; + outcome.dispatched |= plaintext_outcome.dispatched; + outcome.skdm_only |= plaintext_outcome.skdm_only; + } + Err(e) => { + log::warn!( + "[msg:{}] Failed processing plaintext from {}: {e:?}", + info.id, + info.source.sender.observe() + ); + outcome.decrypted = true; + outcome.plaintext_failed = true; + outcome.had_failure = true; + outcome.undecryptable |= + self.handle_plaintext_failure(info, decrypt_fail_mode).await; + } } } Err(e) => { @@ -896,24 +865,33 @@ impl Client { local_identity_reacted = true; self.react_to_local_identity_change(sender_encryption_jid); } - // Same deferral as the main success path. - outcome.decrypted = true; - if defer_dispatch { - pending_dispatch.push(( - enc_type, - decrypted.plaintext, - padding_version, - )); - } else { - self.dispatch_session_plaintext( + let padded_plaintext = decrypted.plaintext; + match self + .clone() + .handle_decrypted_plaintext( enc_type, - &decrypted.plaintext, + &padded_plaintext, padding_version, info, - decrypt_fail_mode, - &mut outcome, ) - .await; + .await + { + Ok(plaintext_outcome) => { + outcome.decrypted = true; + outcome.dispatched |= plaintext_outcome.dispatched; + outcome.skdm_only |= plaintext_outcome.skdm_only; + } + Err(e) => { + log::warn!( + "Failed processing plaintext after identity retry: {e:?}" + ); + outcome.decrypted = true; + outcome.plaintext_failed = true; + outcome.had_failure = true; + outcome.undecryptable |= self + .handle_plaintext_failure(info, decrypt_fail_mode) + .await; + } } } Err(retry_err) => { @@ -1204,30 +1182,6 @@ impl Client { } } } - - // All pairwise ratchet mutations are done. Release the per-sender session - // lock BEFORE dispatching the decrypted plaintext: dispatch (decode, - // SKDM/key-share handling, durable commit, durability hook, synchronous - // event handlers) does not touch the session ratchet the lock guards — - // PASS 2's group path already runs handle_decrypted_plaintext with no - // session lock. Releasing here means a concurrent same-sender op (a - // retry-receipt session rebuild, a reg-id / base-key session reset) no - // longer waits behind this batch's commit + app-controlled durability hook - // + event dispatch. The batch held the lock continuously across every - // decrypt, so no new inter-payload interleaving is introduced. (The crypto - // timer `_t` drops with the function, spanning dispatch as it did before.) - drop(session_guard.take()); - for (enc_type, padded_plaintext, padding_version) in pending_dispatch { - self.dispatch_session_plaintext( - enc_type, - &padded_plaintext, - padding_version, - info, - decrypt_fail_mode, - &mut outcome, - ) - .await; - } outcome } From 8e339d91b58f0f356720f0df2b635f568dfd9c83 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 06:35:00 +0000 Subject: [PATCH 23/30] fix(media): serialize batched reuploads that share a msg_id MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit request_many registered a mediaretry waiter per item keyed on msg_id alone. resolve_waiters wakes every matching waiter with the same notification node, so two batch items sharing a msg_id would cross-resolve — the second decrypts the first's payload with the wrong key. Group items by msg_id so same-id items run sequentially (their waiters never coexist) while distinct ids still fan out concurrently; the all-unique common case is unchanged. --- src/features/media_reupload.rs | 56 +++++++++++++++++++++++++--------- 1 file changed, 42 insertions(+), 14 deletions(-) diff --git a/src/features/media_reupload.rs b/src/features/media_reupload.rs index f4c770066..8d7d2a628 100644 --- a/src/features/media_reupload.rs +++ b/src/features/media_reupload.rs @@ -155,32 +155,60 @@ impl<'a> MediaReupload<'a> { /// Request reupload for several messages at once, concurrently. /// - /// Each request registers its own notification waiter (keyed by the unique - /// message id) and awaits it independently, so a bulk recovery — e.g. many - /// expired-URL media after a long offline period — is bounded by roughly one - /// [`MEDIA_RETRY_TIMEOUT`] instead of the serial sum of per-item waits. - /// Results are returned in the same order as `reqs`; each entry carries that - /// item's success or error (one item failing never aborts the others). + /// Each request registers its own notification waiter and awaits it + /// independently, so a bulk recovery — e.g. many expired-URL media after a + /// long offline period — is bounded by roughly one [`MEDIA_RETRY_TIMEOUT`] + /// instead of the serial sum of per-item waits. Results are returned in the + /// same order as `reqs`; each entry carries that item's success or error + /// (one item failing never aborts the others). + /// + /// The `mediaretry` waiter filters on message id alone, and `resolve_waiters` + /// wakes *every* matching waiter with the same notification node — so two + /// items sharing a `msg_id` would cross-resolve (the second decrypts the + /// first's payload with the wrong key). Items are therefore grouped by + /// `msg_id`: same-id items run sequentially (their waiters never coexist) + /// while distinct ids still fan out concurrently. In the normal case (all + /// ids unique) every item gets its own lane and this is a plain fan-out. pub async fn request_many( &self, reqs: &[MediaReuploadRequest<'_>], ) -> Vec> { use futures::StreamExt; + use std::collections::HashMap; if reqs.is_empty() { return Vec::new(); } - // Stream over owned indices (not a borrow of `reqs` through the - // combinator) and index inside each task, so the fan-out future stays - // Send; collect (index, result) then restore input order. - let mut indexed: Vec<(usize, Result)> = - futures::stream::iter(0..reqs.len()) - .map(|i| async move { (i, self.request(&reqs[i]).await) }) + let mut lanes: HashMap<&str, Vec> = HashMap::new(); + for (i, req) in reqs.iter().enumerate() { + lanes.entry(req.msg_id).or_default().push(i); + } + + // Each lane owns its index list (not a borrow of `reqs` through the + // combinator) and indexes `reqs` inside the task, keeping the fan-out + // future Send. Lanes run concurrently; indices within a lane run in order. + let lane_results: Vec)>> = + futures::stream::iter(lanes.into_values()) + .map(|indices| async move { + let mut out = Vec::with_capacity(indices.len()); + for i in indices { + out.push((i, self.request(&reqs[i]).await)); + } + out + }) .buffer_unordered(MEDIA_REUPLOAD_CONCURRENCY) .collect() .await; - indexed.sort_by_key(|(i, _)| *i); - indexed.into_iter().map(|(_, res)| res).collect() + + let mut results: Vec>> = + (0..reqs.len()).map(|_| None).collect(); + for (i, res) in lane_results.into_iter().flatten() { + results[i] = Some(res); + } + results + .into_iter() + .map(|res| res.expect("every index belongs to exactly one lane")) + .collect() } } From a3c0ff7ebf16aafa4eae47718d62b1b11f4ea5dd Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 06:35:05 +0000 Subject: [PATCH 24/30] test(media): cover intra-attempt host failover in download_media_with_retry MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The existing retry test only exercised the auth-refresh path (single-host conns), so the multi-host fallthrough — a generic error on one host moving to the next in the same attempt, and last_err propagation when all hosts fail — was untested despite the doc comment claiming host-failover coverage. Add both cases. --- src/download.rs | 133 ++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 133 insertions(+) diff --git a/src/download.rs b/src/download.rs index 4c8d88d5f..af07bb0f8 100644 --- a/src/download.rs +++ b/src/download.rs @@ -707,6 +707,139 @@ mod tests { assert!(seen_urls[1].contains("auth=fresh-auth")); } + // A generic (non-auth, non-404) error on one host must fall through to the + // next host within the SAME attempt — no media-conn refresh — and succeed. + #[tokio::test] + async fn download_fails_over_to_next_host_without_refresh() { + let body = b"failover me".to_vec(); + let downloadable = PlaintextDownloadable { + direct_path: "/v/t62.7118-24/failover".to_string(), + file_sha256: plaintext_sha256(&body), + }; + let conn = media_conn( + "auth-tok", + &["bad-host.example.com", "good-host.example.com"], + ); + let refresh_calls = Arc::new(Mutex::new(Vec::new())); + let invalidations = Arc::new(Mutex::new(0usize)); + let seen_urls = Arc::new(Mutex::new(Vec::new())); + + let downloaded = download_media_with_retry( + { + let refresh_calls = Arc::clone(&refresh_calls); + let downloadable = &downloadable; + let conn = conn.clone(); + move |force| { + let refresh_calls = Arc::clone(&refresh_calls); + let conn = conn.clone(); + async move { + refresh_calls.lock().await.push(force); + DownloadUtils::prepare_download_requests( + downloadable, + &wacore::download::MediaConnection::from(&conn), + ) + } + } + }, + { + let invalidations = Arc::clone(&invalidations); + move || { + let invalidations = Arc::clone(&invalidations); + async move { + *invalidations.lock().await += 1; + } + } + }, + { + let seen_urls = Arc::clone(&seen_urls); + let body = body.clone(); + move |request| { + let seen_urls = Arc::clone(&seen_urls); + let body = body.clone(); + let url = request.url.clone(); + async move { + seen_urls.lock().await.push(url.clone()); + if url.contains("bad-host") { + Err(DownloadRequestError::other(anyhow!("connection reset"))) + } else { + Ok(body) + } + } + } + }, + ) + .await + .expect("download should fail over to the healthy host"); + + assert_eq!(downloaded, body); + // Single attempt, no refresh: a generic error doesn't invalidate the media conn. + assert_eq!(*refresh_calls.lock().await, vec![false]); + assert_eq!(*invalidations.lock().await, 0); + let seen_urls = seen_urls.lock().await.clone(); + assert_eq!(seen_urls.len(), 2); + assert!(seen_urls[0].contains("bad-host")); + assert!(seen_urls[1].contains("good-host")); + } + + // When every host fails with a generic error, the accumulated `last_err` + // is surfaced (not the fallback "all hosts" message) and no refresh happens. + #[tokio::test] + async fn download_propagates_last_error_when_all_hosts_fail() { + let body = b"never arrives".to_vec(); + let downloadable = PlaintextDownloadable { + direct_path: "/v/t62.7118-24/allfail".to_string(), + file_sha256: plaintext_sha256(&body), + }; + let conn = media_conn("auth-tok", &["host-a.example.com", "host-b.example.com"]); + let invalidations = Arc::new(Mutex::new(0usize)); + let seen_urls = Arc::new(Mutex::new(Vec::new())); + + let err = download_media_with_retry( + { + let downloadable = &downloadable; + let conn = conn.clone(); + move |_force| { + let conn = conn.clone(); + async move { + DownloadUtils::prepare_download_requests( + downloadable, + &wacore::download::MediaConnection::from(&conn), + ) + } + } + }, + { + let invalidations = Arc::clone(&invalidations); + move || { + let invalidations = Arc::clone(&invalidations); + async move { + *invalidations.lock().await += 1; + } + } + }, + { + let seen_urls = Arc::clone(&seen_urls); + move |request| { + let seen_urls = Arc::clone(&seen_urls); + let url = request.url.clone(); + async move { + seen_urls.lock().await.push(url.clone()); + Err::, _>(DownloadRequestError::other(anyhow!("host {url} down"))) + } + } + }, + ) + .await + .expect_err("all hosts failing must surface an error"); + + assert!( + err.to_string().contains("down"), + "expected the propagated last_err, got: {err}" + ); + assert_eq!(*invalidations.lock().await, 0); + assert_eq!(seen_urls.lock().await.len(), 2); + } + #[tokio::test] async fn download_to_writer_retries_with_forced_media_conn_refresh_after_auth_error() { let body = b"stream me".to_vec(); From df189394445b4e97c981490d28e9bac8f73d705c Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 06:42:06 +0000 Subject: [PATCH 25/30] fix(media): reject duplicate msg_ids in request_many instead of serializing MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Serializing same-id items still left a late-response window: once the first waiter times out it lingers in node_waiters (canceled waiters are purged only lazily by resolve_waiters), so its delayed mediaretry could resolve a same-id retry with the wrong payload. A message id is unique per message, so a duplicate in a batch is a caller mistake — reject it (past the first occurrence) with InvalidRequest so no two waiters ever share an id filter, timeout or not. --- src/features/media_reupload.rs | 57 ++++++++++++++++++---------------- 1 file changed, 30 insertions(+), 27 deletions(-) diff --git a/src/features/media_reupload.rs b/src/features/media_reupload.rs index 8d7d2a628..d6410d4b3 100644 --- a/src/features/media_reupload.rs +++ b/src/features/media_reupload.rs @@ -162,52 +162,55 @@ impl<'a> MediaReupload<'a> { /// same order as `reqs`; each entry carries that item's success or error /// (one item failing never aborts the others). /// - /// The `mediaretry` waiter filters on message id alone, and `resolve_waiters` - /// wakes *every* matching waiter with the same notification node — so two - /// items sharing a `msg_id` would cross-resolve (the second decrypts the - /// first's payload with the wrong key). Items are therefore grouped by - /// `msg_id`: same-id items run sequentially (their waiters never coexist) - /// while distinct ids still fan out concurrently. In the normal case (all - /// ids unique) every item gets its own lane and this is a plain fan-out. + /// Duplicate `msg_id`s in one batch are rejected (past the first occurrence) + /// with [`MediaReuploadError::InvalidRequest`]. The `mediaretry` waiter + /// filters on message id alone and `resolve_waiters` wakes *every* match, so + /// two same-id waiters would cross-resolve. Serializing them isn't enough: + /// once the first waiter times out it lingers in `node_waiters` (canceled + /// waiters are purged only lazily), so its late notification could still + /// resolve a same-id retry with the wrong payload. Requiring unique ids means + /// no two waiters ever share a filter — a message id is unique per message, + /// so a duplicate is a caller mistake, not a real recovery target. pub async fn request_many( &self, reqs: &[MediaReuploadRequest<'_>], ) -> Vec> { use futures::StreamExt; - use std::collections::HashMap; + use std::collections::HashSet; if reqs.is_empty() { return Vec::new(); } - let mut lanes: HashMap<&str, Vec> = HashMap::new(); + let mut results: Vec>> = + (0..reqs.len()).map(|_| None).collect(); + let mut seen: HashSet<&str> = HashSet::with_capacity(reqs.len()); + let mut unique: Vec = Vec::with_capacity(reqs.len()); for (i, req) in reqs.iter().enumerate() { - lanes.entry(req.msg_id).or_default().push(i); + if seen.insert(req.msg_id) { + unique.push(i); + } else { + results[i] = Some(Err(MediaReuploadError::InvalidRequest(format!( + "duplicate msg_id {} in batch", + req.msg_id + )))); + } } - // Each lane owns its index list (not a borrow of `reqs` through the - // combinator) and indexes `reqs` inside the task, keeping the fan-out - // future Send. Lanes run concurrently; indices within a lane run in order. - let lane_results: Vec)>> = - futures::stream::iter(lanes.into_values()) - .map(|indices| async move { - let mut out = Vec::with_capacity(indices.len()); - for i in indices { - out.push((i, self.request(&reqs[i]).await)); - } - out - }) + // Stream over owned indices (not a borrow of `reqs` through the + // combinator) and index inside each task, so the fan-out future stays + // Send. Every id here is unique, so no two waiters share a filter. + let done: Vec<(usize, Result)> = + futures::stream::iter(unique) + .map(|i| async move { (i, self.request(&reqs[i]).await) }) .buffer_unordered(MEDIA_REUPLOAD_CONCURRENCY) .collect() .await; - - let mut results: Vec>> = - (0..reqs.len()).map(|_| None).collect(); - for (i, res) in lane_results.into_iter().flatten() { + for (i, res) in done { results[i] = Some(res); } results .into_iter() - .map(|res| res.expect("every index belongs to exactly one lane")) + .map(|res| res.expect("every index is either a duplicate or fetched")) .collect() } } From 9e7e15facb992d9234d7f5810a820d0d14c279b4 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 13:36:21 +0000 Subject: [PATCH 26/30] fix(media): bound the streaming download reader to max_body_bytes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit execute() caps read_to_vec at max_body_bytes (2 GiB, WA's max file size), but execute_streaming returned an unbounded reader — fine when the caller owns the sink, but download() now streams into an in-memory Vec, so a CDN that streams past the declared length could grow it until OOM (DOWNLOAD_PREALLOC_CAP only sizes the initial allocation). Cap the streaming reader with the same max_body_bytes: over the cap it hits EOF and the downstream MAC/SHA check rejects it. Restores the pre-refactor in-memory bound. --- http_clients/ureq-client/src/lib.rs | 44 +++++++++++++++++++++++++---- 1 file changed, 39 insertions(+), 5 deletions(-) diff --git a/http_clients/ureq-client/src/lib.rs b/http_clients/ureq-client/src/lib.rs index 0c2a87a4b..6fa06410e 100644 --- a/http_clients/ureq-client/src/lib.rs +++ b/http_clients/ureq-client/src/lib.rs @@ -23,8 +23,9 @@ const MAX_IDLE_CONNECTIONS: u64 = 3; #[derive(Debug, Clone)] pub struct UreqHttpClient { agent: ureq::Agent, - /// Cap for [`UreqHttpClient::execute`]. Streaming is unbounded — the - /// caller owns the sink. + /// Total-bytes cap for both [`UreqHttpClient::execute`] and the reader from + /// [`UreqHttpClient::execute_streaming`]. Bounds an in-memory sink so a + /// hostile CDN can't drive it to OOM; defaults to WA's 2 GiB max file size. max_body_bytes: u64, /// Best-effort pool footprint for `resource_report`. `None` when a custom /// agent is supplied (its buffer/pool config is opaque to us). @@ -64,8 +65,9 @@ impl UreqHttpClient { } } - /// Override the per-response cap for [`UreqHttpClient::execute`]. Set to - /// `u64::MAX` to disable; a hostile server can then exhaust memory. + /// Override the per-response cap for [`UreqHttpClient::execute`] and + /// [`UreqHttpClient::execute_streaming`]. Set to `u64::MAX` to disable; a + /// hostile server can then exhaust memory. pub fn with_max_body_bytes(mut self, max_body_bytes: u64) -> Self { self.max_body_bytes = max_body_bytes; self @@ -172,7 +174,13 @@ impl HttpClient for UreqHttpClient { }; let status_code = response.status().as_u16(); - let reader = response.into_body().into_reader(); + // Bound the streaming reader to the same cap `execute` enforces: an + // in-memory sink (`Client::download` buffers into a `Vec`) must not be + // driveable to OOM by a CDN that streams past the declared length. Over + // the cap the reader hits EOF and the downstream MAC/SHA check fails, + // rather than growing the sink unbounded. `DOWNLOAD_PREALLOC_CAP` only + // sizes the initial allocation, not the total read. + let reader = std::io::Read::take(response.into_body().into_reader(), self.max_body_bytes); Ok(StreamingHttpResponse { status_code, @@ -302,6 +310,32 @@ mod tests { .expect_err("1 KiB cap must reject a 4 MiB body"); } + // The streaming reader must honor the same cap: an over-cap body is + // truncated at EOF (the caller's decrypt/MAC check then rejects it) instead + // of growing an in-memory sink to OOM. + #[tokio::test(flavor = "current_thread")] + async fn execute_streaming_bounds_body_at_cap() { + const SIZE: usize = 4 * 1024 * 1024; + const CAP: u64 = 1024; + let url = spawn_fixed_size_server(SIZE); + let read = tokio::task::spawn_blocking(move || { + let mut resp = UreqHttpClient::new() + .with_max_body_bytes(CAP) + .execute_streaming(HttpRequest { + method: "GET".into(), + url, + headers: std::collections::HashMap::new(), + body: None, + }) + .expect("streaming GET should start"); + let mut sink = std::io::sink(); + std::io::copy(&mut resp.body, &mut sink).expect("draining the reader should not error") + }) + .await + .unwrap(); + assert_eq!(read, CAP, "streaming body must stop at the cap"); + } + /// Captures the raw request headers and body of a single POST, then replies 200. fn spawn_capture_server() -> (String, std::sync::mpsc::Receiver<(String, Vec)>) { let listener = TcpListener::bind("127.0.0.1:0").expect("bind ephemeral port"); From 59d67f74f3e67e6234f4ede0b93275c213eaad03 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 13:46:00 +0000 Subject: [PATCH 27/30] fix(connect): order login pre-key upload before signed-pre-key rotation Detaching the login pre-key upload (so it no longer gates set_passive) let it overlap RotateKeyJob. Both re-declare the signed pre-key to the server: the upload bundles the CURRENT one (from its snapshot) with the one-time keys, rotation uploads a freshly promoted one. If rotation lands first, the upload reverts the server to the stale signed pre-key; once that key is pruned locally, pkmsg sessions the server hands out become undecryptable. Run both on one detached task in order (upload -> rotate) so set_passive stays un-gated while rotation reads the upload's persisted state. --- src/client/node_io.rs | 44 +++++++++++++++++++------------------------ 1 file changed, 19 insertions(+), 25 deletions(-) diff --git a/src/client/node_io.rs b/src/client/node_io.rs index 1b954f2f1..17e73e037 100644 --- a/src/client/node_io.rs +++ b/src/client/node_io.rs @@ -748,44 +748,38 @@ impl Client { // (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. + // Pre-key upload then RotateKeyJob, ordered on ONE detached task. + // Both re-declare the signed pre-key to the server — the upload bundles + // the CURRENT one with its one-time keys, rotation uploads a freshly + // promoted one. Run as two independent tasks they can overlap, and if + // rotation lands first, the upload (built from a pre-rotation snapshot) + // reverts the server to the stale signed pre-key; once that key is + // pruned, pkmsg sessions the server hands out become undecryptable. + // Ordering them here keeps set_passive un-gated (still detached) while + // making rotation read the upload's persisted state. check_generation!(); - let prekey_client = client_clone.clone(); - let prekey_generation = task_generation; + let key_client = client_clone.clone(); + let key_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 - { + if key_client.connection_generation.load(Ordering::SeqCst) != key_generation { return; } - if let Err(e) = prekey_client.upload_pre_keys_at_login().await - && !prekey_client.is_shutting_down() + if let Err(e) = key_client.upload_pre_keys_at_login().await + && !key_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 - // post-login init. - check_generation!(); - let rotate_client = client_clone.clone(); - let rotate_generation = task_generation; - client_clone - .runtime - .spawn(Box::pin(async move { - // A newer connection may have taken over between spawn and now; - // rotating on a stale generation would upload a duplicate key. - if rotate_client.connection_generation.load(Ordering::SeqCst) - != rotate_generation - { + // The upload awaited network I/O; re-check before rotating so a + // stale generation doesn't upload a duplicate signed pre-key. + if key_client.connection_generation.load(Ordering::SeqCst) != key_generation { return; } - if let Err(e) = rotate_client.maybe_rotate_signed_pre_key().await - && !rotate_client.is_shutting_down() + if let Err(e) = key_client.maybe_rotate_signed_pre_key().await + && !key_client.is_shutting_down() { warn!("Signed pre-key rotation check failed: {e:?}"); } From 3fcb3f5986d30a15b2b521f6c891b200fa159d68 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 14:01:08 +0000 Subject: [PATCH 28/30] fix(history-sync): serialize tc-token get-then-store across concurrent chunks Ingesting history-sync chunks concurrently (bounded) removed the serial worker's implicit ordering around the tc-token write. store_tc_token_candidate guards newer-wins with a non-atomic get_tc_token read then an unconditional store_received_tc_token, so two same-contact candidates in different chunks could read the same baseline and let the older one's write land last, clobbering a fresher privacy token. Serialize the read-check-write on a per-client tc_token_lock, restoring the pre-parallelization ordering without touching the shared store semantics the privacy-notification path depends on. --- src/client.rs | 5 +++++ src/client/lifecycle.rs | 1 + src/history_sync.rs | 6 ++++++ 3 files changed, 12 insertions(+) diff --git a/src/client.rs b/src/client.rs index bbd3461be..7b8b115d7 100644 --- a/src/client.rs +++ b/src/client.rs @@ -646,6 +646,11 @@ pub struct Client { /// Single-flights signed pre-key rotation so overlapping post-login tasks /// (from reconnect churn) can't run the rotate/upload/prune flow concurrently. pub(crate) signed_pre_key_rotation_lock: Arc>, + /// Serializes the history-sync tc-token get-then-store: chunks ingest + /// concurrently, but the newer-wins guard is a non-atomic read-then-write, + /// so without this two same-contact candidates could interleave and let an + /// older privacy token overwrite a fresher one. + pub(crate) tc_token_lock: Arc>, /// Notifier for when offline sync (ib offline stanza) is received. /// WhatsApp Web waits for this before sending passive tasks (prekey upload, active IQ, presence). pub(crate) offline_sync_notifier: Arc, diff --git a/src/client/lifecycle.rs b/src/client/lifecycle.rs index d2c34c4e7..a86efe357 100644 --- a/src/client/lifecycle.rs +++ b/src/client/lifecycle.rs @@ -218,6 +218,7 @@ impl Client { initial_app_state_keys_received: Arc::new(AtomicBool::new(false)), prekey_upload_lock: Arc::new(async_lock::Mutex::new(())), signed_pre_key_rotation_lock: Arc::new(async_lock::Mutex::new(())), + tc_token_lock: Arc::new(async_lock::Mutex::new(())), offline_sync_notifier: Arc::new(event_listener::Event::new()), offline_sync_completed: Arc::new(AtomicBool::new(false)), offline_sync_finish_started: Arc::new(AtomicBool::new(false)), diff --git a/src/history_sync.rs b/src/history_sync.rs index 0d1294a49..685ecf464 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -407,6 +407,12 @@ impl Client { let backend = self.persistence_manager.backend(); + // Serialize the get-then-store below: history-sync chunks ingest + // concurrently, so without this two same-contact candidates could + // read the same baseline and the older one's unconditional write could + // land last, clobbering a fresher privacy token. + let _guard = self.tc_token_lock.lock().await; + // Skip only when an existing *real* token is newer than this candidate; a // byte-less placeholder stamps token_timestamp with a sender epoch and // must never block the first real token from history sync. From 0a9af660bdc9e1e1653f00c7294cf6994eea695d Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 14:33:11 +0000 Subject: [PATCH 29/30] docs(sync): clarify the major-sync channel carries only history tasks MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The recv loop's HistorySync-vs-else shape read as though app-state patches flow through this channel and could be head-of-line blocked by the history semaphore. They don't: nothing enqueues MajorSyncTask::AppStateSync — app-state sync runs via its own direct path (fetch_app_state_with_retry). Document that the channel carries only HistorySync tasks (so the permit backpressures history intake and can't delay app-state), mark the else arm defensive, and drop the stale 'queued task holds a decompressed blob' note (queued tasks hold only the notification; only in-flight chunks decompress). --- src/bot.rs | 17 ++++++++++------- 1 file changed, 10 insertions(+), 7 deletions(-) diff --git a/src/bot.rs b/src/bot.rs index 48dd416bb..dd6523a09 100644 --- a/src/bot.rs +++ b/src/bot.rs @@ -467,12 +467,14 @@ 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. + // This channel carries only HistorySync tasks: app-state sync runs via + // its own direct path (fetch_app_state_with_retry), nothing enqueues + // AppStateSync here. Chunks are independent (order-free upserts; the + // event carries chunk_order), so ingest concurrently, bounded low — each + // in-flight chunk decompresses a blob and the connect path is peak- + // memory-conscious (WA Web caps at histSyncChunk=3). Taking the permit in + // the recv loop backpressures history intake on a burst; since no + // app-state task flows here, that can't head-of-line block one. const HISTORY_SYNC_CONCURRENCY: usize = 2; let worker_client = Arc::downgrade(&client); let history_permits = Arc::new(async_lock::Semaphore::new(HISTORY_SYNC_CONCURRENCY)); @@ -495,7 +497,8 @@ impl Bot { })) .detach(); } else { - // App-state sync: serial + ordered. + // Defensive: nothing enqueues AppStateSync today, but if + // that changes it must run serially (ordered patches). worker_client.process_sync_task(task).await; } } From 89fe09fea8065637907834303bb40f40594e4398 Mon Sep 17 00:00:00 2001 From: Claude Date: Sat, 4 Jul 2026 14:52:33 +0000 Subject: [PATCH 30/30] docs(history-sync): note the in-flight counter's underflow clamp finish_history_sync_task's 'previous <= 1 -> store(0)' branch already absorbs a late finish after cleanup_connection_state() zeroed the counter: fetch_sub from 0 momentarily wraps to usize::MAX, but previous == 0 takes the branch and stores 0, so the wrap never sticks and the idle waiter is never left blocked. Document that so the fetch_sub isn't misread as an unguarded decrement (no behavior change). --- src/client/sessions.rs | 5 +++++ 1 file changed, 5 insertions(+) diff --git a/src/client/sessions.rs b/src/client/sessions.rs index a080d11b2..470dece77 100644 --- a/src/client/sessions.rs +++ b/src/client/sessions.rs @@ -219,6 +219,11 @@ impl Client { } pub(crate) fn finish_history_sync_task(&self) { + // The `previous <= 1` clamp is also the underflow guard: a detached task + // that finishes after cleanup_connection_state() reset the counter to 0 + // hits fetch_sub-from-0, which momentarily wraps the stored value to + // usize::MAX — but previous == 0 takes this branch and stores 0, so the + // wrap never sticks (and it never leaves the idle waiter blocked). let previous = self .history_sync_tasks_in_flight .fetch_sub(1, Ordering::Relaxed);