perf: parallelize remaining serial/blocking hot paths (startup, media, send/recv) - #975
Conversation
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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0116cfRpumbEv3H4qUNmtNk4
`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 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_0116cfRpumbEv3H4qUNmtNk4
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).
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.
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).
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.
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.
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.
…erial 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.
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).
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.
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).
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.
📝 WalkthroughWalkthroughThis PR replaces several sequential await chains with bounded concurrent execution across sync, login, lookup, upload, and download paths, and rewrites media downloads to stream into a writer with capped preallocation. ChangesBounded concurrency and streaming download refactor
Estimated code review effort: 4 (Complex) | ~60 minutes Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 368795fef7
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
📦 Binary size report
.text per crate
Top movers (cargo-bloat attribution)
Baseline: |
|
| Filename | Overview |
|---|---|
| src/download.rs | download() streams each attempt into a fresh Cursor (stale-tail safe, overlaps network+decrypt); three new unit tests cover host-failover, all-hosts-fail, and auth-refresh. |
| src/bot.rs | History-sync intake loop fans out chunks behind a cap-2 semaphore; shutdown log message corrected to reflect detached tasks may still be running. |
| src/client/app_state.rs | Serial blob pre-download refactored into concurrent fan-out with directPath dedup; addresses the previously-flagged duplicate-download concern. |
| http_clients/ureq-client/src/lib.rs | execute_streaming now applies max_body_bytes via Read::take; test verifies truncation at the cap. |
| src/history_sync.rs | tc_token_lock serializes the non-atomic get-check-write; lock correctly scoped after LID resolution and before the read-then-write window. |
| src/features/media_reupload.rs | request_many adds batched concurrent reupload; duplicate msg_id entries rejected; index-preserving result reassembly is sound. |
| src/client/node_io.rs | Pre-key upload detached from set_passive; generation re-checked after network I/O so stale rotations don't clobber the freshly persisted signed pre-key. |
| src/features/contacts.rs | join! avoids leaking response_waiters entry on cancellation; both PN and LID IQ results awaited to completion before error propagation. |
| src/send/mod.rs | Status LID resolution fanned out with buffer_unordered(16); index-preservation correct via pre-sized vec![None; n] + indexed assignment. |
| src/usync.rs | Local registry scan parallelized with buffer_unordered(16); None results fall through to the network usync IQ as before. |
| src/client/sessions.rs | has_session probes run concurrently; underflow guard in finish_history_sync_task correctly documented. |
| src/prekeys.rs | Companion identity loads fanned out concurrently; owned Vec prevents borrow-through-combinator; filter_map drops None entries correctly. |
| src/message/retry.rs | Two independent JID resolution calls parallelized with futures::join!; minimal change with no behavioral risk. |
| wacore/src/sticker_pack.rs | Doc example updated to concurrent zip+thumbnail upload pattern; no library code changed. |
| src/client.rs | Adds tc_token_lock field to Client. |
| src/client/lifecycle.rs | Initializes tc_token_lock alongside the other async_lock::Mutex fields. |
Flowchart
%%{init: {'theme': 'neutral'}}%%
flowchart TD
subgraph Startup
A[set_passive gate] -->|detached task| B[upload_pre_keys_at_login]
B -->|check gen| C[maybe_rotate_signed_pre_key]
D[App-state IQ response] --> E{pre_download_external_blobs}
E -->|buffer_unordered 4| F[CDN GET blob 1..N dedup by directPath]
F --> G[Process patch lists]
end
subgraph HistorySync
H[channel recv] --> I{HistorySync?}
I -- yes --> J[acquire semaphore cap=2]
J --> K[spawn detached task]
K -->|tc_token_lock serialize read-check-write| L[(tc_token store)]
I -- no --> M[process serially]
end
subgraph Media
N[download] --> O[fresh Cursor per attempt]
O --> P[streaming_download_and_decrypt spawn_blocking]
P -- auth/404 --> Q[invalidate_media_conn next attempt]
P -- generic err --> R[next host same attempt]
P -- ok --> S[Vec into_inner]
end
subgraph Batch
T[request_many] -->|dedup msg_id| U[buffer_unordered 32]
V[get_device_lists] -->|buffer_unordered 16| W[registry reads]
X[status recipients] -->|buffer_unordered 16| Y[resolve_to_lid]
Y -->|rebuild by index| Z[resolved Vec ordered]
end
%%{init: {'theme': 'base', 'themeVariables': {"darkMode": true, "background": "#0d1117", "primaryColor": "#21262d", "primaryTextColor": "#e6edf3", "primaryBorderColor": "#8b949e", "lineColor": "#8b949e", "textColor": "#e6edf3", "edgeLabelBackground": "#161b22", "actorBkg": "#21262d", "actorBorder": "#8b949e", "actorTextColor": "#e6edf3", "actorLineColor": "#8b949e", "signalColor": "#8b949e", "signalTextColor": "#e6edf3", "noteBkgColor": "#373320", "noteBorderColor": "#d4a72c", "noteTextColor": "#f0e6c0", "labelBoxBkgColor": "#21262d", "labelBoxBorderColor": "#8b949e", "labelTextColor": "#e6edf3", "loopTextColor": "#e6edf3", "activationBkgColor": "#30363d", "activationBorderColor": "#8b949e"}}}%%
flowchart TD
subgraph Startup
A[set_passive gate] -->|detached task| B[upload_pre_keys_at_login]
B -->|check gen| C[maybe_rotate_signed_pre_key]
D[App-state IQ response] --> E{pre_download_external_blobs}
E -->|buffer_unordered 4| F[CDN GET blob 1..N dedup by directPath]
F --> G[Process patch lists]
end
subgraph HistorySync
H[channel recv] --> I{HistorySync?}
I -- yes --> J[acquire semaphore cap=2]
J --> K[spawn detached task]
K -->|tc_token_lock serialize read-check-write| L[(tc_token store)]
I -- no --> M[process serially]
end
subgraph Media
N[download] --> O[fresh Cursor per attempt]
O --> P[streaming_download_and_decrypt spawn_blocking]
P -- auth/404 --> Q[invalidate_media_conn next attempt]
P -- generic err --> R[next host same attempt]
P -- ok --> S[Vec into_inner]
end
subgraph Batch
T[request_many] -->|dedup msg_id| U[buffer_unordered 32]
V[get_device_lists] -->|buffer_unordered 16| W[registry reads]
X[status recipients] -->|buffer_unordered 16| Y[resolve_to_lid]
Y -->|rebuild by index| Z[resolved Vec ordered]
end
Reviews (10): Last reviewed commit: "docs(history-sync): note the in-flight c..." | Re-trigger Greptile
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/download.rs (1)
276-303: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winDocument the empty-writer contract Rewinding to
0does not clear stale tail bytes, so a reusedCursor<Vec<u8>>orFilecan return extra bytes from a previous write even though the plaintext hash still matches. Existing callers pass empty cursors, but this public API still needs a hard contract or truncation where possible.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/download.rs` around lines 276 - 303, Clarify and enforce the empty-writer contract for download_to_writer: the current seek-to-start behavior in download_to_writer/download_to_writer_with_retry does not remove stale tail bytes when a reused W or Cursor<Vec<u8>> is longer than the decrypted payload. Update the docs for download_to_writer (and any related helper like streaming_download_and_decrypt if needed) to state that callers must provide an empty, writable target, or add truncation/length-reset logic where the writer type supports it before rewinding to 0.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/client/app_state.rs`:
- Around line 83-119: The predownload flow in pre_download_external_blobs() is
still retaining all downloaded blob bytes in the HashMap until the end, causing
high peak memory and extra Vec<u8> cloning during the inline passes. Update the
app_state.rs flow so blobs are consumed incrementally instead of kept resident
for the full batch, either by draining entries from the pre_downloaded map as
they are used or by inlining processing directly from the download stream. Keep
the change localized around the pre_download_external_blobs() logic and the
callers that read from the pre_downloaded cache.
In `@src/features/contacts.rs`:
- Around line 148-180: The concurrent PN/LID lookup in the is_on_whatsapp flow
currently uses futures::join!, which waits for both IsOnWhatsAppSpec requests to
finish even if one errors; update this block to fail fast by switching to a
try_join-style approach so an error from either pn_fut or lid_fut returns
immediately and the other in-flight future is dropped. Keep the existing pn_fut
and lid_fut structure, and preserve the final result merging logic after both
succeed.
In `@src/message/receive.rs`:
- Around line 1147-1182: Post-decrypt dispatch in process_session_enc_batch can
run concurrently across chats for the same sender because chat_lanes are
per-chat, which allows SKDM/key-share handling, durable commits, and event
dispatch to be reordered. Fix this by adding a sender-scoped dispatch lock or
enforcing a stronger upstream same-sender serialization guarantee around the
handle_decrypted_plaintext loop, using the existing
session_guard/pending_dispatch flow as the place to preserve sender order.
---
Outside diff comments:
In `@src/download.rs`:
- Around line 276-303: Clarify and enforce the empty-writer contract for
download_to_writer: the current seek-to-start behavior in
download_to_writer/download_to_writer_with_retry does not remove stale tail
bytes when a reused W or Cursor<Vec<u8>> is longer than the decrypted payload.
Update the docs for download_to_writer (and any related helper like
streaming_download_and_decrypt if needed) to state that callers must provide an
empty, writable target, or add truncation/length-reset logic where the writer
type supports it before rewinding to 0.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: db0ba7ff-6cb8-44ea-be25-58b149601891
📒 Files selected for processing (13)
src/bot.rssrc/client/app_state.rssrc/client/node_io.rssrc/client/sessions.rssrc/download.rssrc/features/contacts.rssrc/features/media_reupload.rssrc/message/receive.rssrc/message/retry.rssrc/prekeys.rssrc/send/mod.rssrc/usync.rswacore/src/sticker_pack.rs
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
3 issues found across 13 files
Confidence score: 2/5
- In
src/download.rs, routingdownload()throughdownload_to_writer()with a sharedCursor<Vec<u8>>can leave stale bytes after a seek on retry/failover, which risks returning corrupted download data to users — reset/truncate the buffer (or allocate a fresh writer per retry path) before merging. - In
src/message/receive.rs, mixed dispatch behavior in the PN→LID migration path can let later payloads deliver before earlier deferred successes, potentially reordering session messages and causing incorrect client state transitions — make migration successes follow the same deferred ordering path (or add explicit ordering guarantees) before merging. - In
src/bot.rs, the shutdown log can report "Sync worker shutting down" while detachedHistorySynctasks are still active, which can mislead ops/debugging during incident handling — either await/cancel detached tasks on shutdown or make the log explicitly indicate background work may continue.
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
Routing download() through the shared-writer path meant retries/host-failover reused one Cursor<Vec<u8>>, 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.
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.
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.
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.
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").
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: edc177c34d
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
There was a problem hiding this comment.
Actionable comments posted: 1
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@src/features/contacts.rs`:
- Around line 178-182: The `try_join!` in `contacts.rs` can drop the in-flight
LID future while leaving its waiter entry behind, so add cancellation-safe
cleanup in the `send_and_wait_iq` path. Update the logic around
`response_waiters` registration/removal so that if the future is canceled after
the waiter is inserted, the entry is always removed even when the request is
abandoned before send success, timeout, or shutdown. Use the `send_and_wait_iq`
function and the `response_waiters` bookkeeping to locate and fix the cleanup
behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: aa4365ba-d2cb-43da-83ae-d84db992e141
📒 Files selected for processing (10)
src/bot.rssrc/client/app_state.rssrc/client/node_io.rssrc/client/sessions.rssrc/download.rssrc/features/contacts.rssrc/message/receive.rssrc/prekeys.rssrc/send/mod.rssrc/usync.rs
There was a problem hiding this comment.
1 issue found across 10 files (changes from recent commits).
Reply with feedback, questions, or to request a fix.
Re-trigger cubic
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.
There was a problem hiding this comment.
0 issues found across 2 files (changes from recent commits).
Auto-approved: Parallelizes serial hot paths (download streaming, app-state blob pre-download, session probing, contact/LID resolver, registry reads) with bounded concurrency. Safe per analysis and tested.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: df18939444
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
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.
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: PR changes major concurrency across critical paths with a high blast radius; bugs could disrupt core messaging or data integrity. With 778 lines touching hot paths, a human review is warranted.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 9e7e15facb
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
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.
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: This PR introduces extensive parallelization and refactoring across startup, media, send/recv, and sync paths, with significant architectural changes to concurrency, data flow, and error handling that require careful human review of correctness and safety.
Re-trigger cubic
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 59d67f74f3
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
…t 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.
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
There was a problem hiding this comment.
0 issues found across 3 files (changes from recent commits).
Requires human review: Core concurrency changes across startup, media, sync, sessions, and device resolution — high risk of race conditions and correctness issues despite careful review.
Re-trigger cubic
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).
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: Non-trivial optimizations to core hot paths (sync, media, session, contacts) with risk of regressions in production; requires human review.
Re-trigger cubic
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
src/bot.rs (1)
489-509: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winPrevent the history-sync counter from wrapping at shutdown
finish_history_sync_task()still does a plainfetch_sub(1), andcleanup_connection_state()resetshistory_sync_tasks_in_flightto0. A detached history-sync task that finishes after cleanup will underflow the counter tousize::MAX, which breaks the idle/wait-for-idle path. Make the decrement saturating or guard the finish path against post-cleanup completion.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/bot.rs` around lines 489 - 509, The history-sync in-flight counter can underflow if a detached task calls finish_history_sync_task() after cleanup_connection_state() has already reset history_sync_tasks_in_flight to zero. Update finish_history_sync_task() to use a saturating or guarded decrement so late completions cannot wrap the counter, and keep the idle/wait-for-idle logic in bot.rs consistent with the task lifecycle around the detached history-sync spawn.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/bot.rs`:
- Around line 489-509: The history-sync in-flight counter can underflow if a
detached task calls finish_history_sync_task() after cleanup_connection_state()
has already reset history_sync_tasks_in_flight to zero. Update
finish_history_sync_task() to use a saturating or guarded decrement so late
completions cannot wrap the counter, and keep the idle/wait-for-idle logic in
bot.rs consistent with the task lifecycle around the detached history-sync
spawn.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 9cb6e378-3ac3-4bfe-a056-34be58ed3f2d
📒 Files selected for processing (7)
http_clients/ureq-client/src/lib.rssrc/bot.rssrc/client.rssrc/client/lifecycle.rssrc/client/node_io.rssrc/features/media_reupload.rssrc/history_sync.rs
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).
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
There was a problem hiding this comment.
0 issues found across 1 file (changes from recent commits).
Requires human review: Significant concurrency refactor across many core paths (startup, media, sync, send). While well-documented, the breadth and impact warrant human review to catch potential race conditions or behavioral changes.
Re-trigger cubic
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/client/sessions.rs (2)
292-332: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueBounded fan-out looks solid — but let's not leave loose numbers lying around.
The
buffer_unordered(16)migration to concurrent session probing is the right call — we don't ship products that make people wait in line one-by-one when they don't have to.Ok(false)/Errhandling correctly funnels intojids_needing_sessions, and using unordered collection since downstream batching doesn't care about order is exactly the kind of pragmatic tradeoff I like to see.One nit:
16is a bare magic number here, whilecrate::session::SESSION_CHECK_BATCH_SIZEis used right below at Line 327 for the follow-up batching. If these two numbers are meant to be independently tunable, a named constant (e.g.,SESSION_PROBE_CONCURRENCY) would make the intent clearer and prevent someone from "optimizing" one without realizing the other exists.♻️ Suggested constant extraction
+const SESSION_PROBE_CONCURRENCY: usize = 16; + async fn ensure_sessions_inner(&self, jids: Vec<Jid>) -> Result<()> { ... - .buffer_unordered(16) + .buffer_unordered(SESSION_PROBE_CONCURRENCY)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/client/sessions.rs` around lines 292 - 332, The concurrent probe limit in ensure_sessions_inner uses a bare magic number for buffer_unordered, which makes the intent and tuning unclear. Replace the literal concurrency value in ensure_sessions_inner with a named constant (for example, one dedicated to session probe concurrency) defined alongside the session-related constants, and use that symbol in the futures::StreamExt pipeline so it is easy to locate and adjust independently from crate::session::SESSION_CHECK_BATCH_SIZE.
448-455: 🚀 Performance & Scalability | 🔵 Trivial | 💤 Low valueTwo independent checks, one at a time — small win, but a win.
lid_existsandpn_existscheck unrelated JIDs and could run concurrently instead of back-to-back.awaits. Given this only runs once at login for the primary phone, the payoff is marginal — I'm flagging it for completeness rather than demanding it, since we shouldn't burn effort chasing every microsecond when the audience for this fix is basically nobody.♻️ Optional concurrent check
- let lid_exists = self - .check_session_exists(&primary_phone_lid) - .await - .unwrap_or(false); - let pn_exists = self - .check_session_exists(&primary_phone_pn) - .await - .unwrap_or(false); + let (lid_exists, pn_exists) = tokio::join!( + self.check_session_exists(&primary_phone_lid), + self.check_session_exists(&primary_phone_pn), + ); + let lid_exists = lid_exists.unwrap_or(false); + let pn_exists = pn_exists.unwrap_or(false);🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@src/client/sessions.rs` around lines 448 - 455, The two independent session-existence checks in the login flow are run sequentially even though they query unrelated JIDs. Update the logic around check_session_exists in src/client/sessions.rs so lid_exists and pn_exists are evaluated concurrently rather than with back-to-back awaits, keeping the same fallback behavior on error and preserving the existing variables used by the login path.
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Outside diff comments:
In `@src/client/sessions.rs`:
- Around line 292-332: The concurrent probe limit in ensure_sessions_inner uses
a bare magic number for buffer_unordered, which makes the intent and tuning
unclear. Replace the literal concurrency value in ensure_sessions_inner with a
named constant (for example, one dedicated to session probe concurrency) defined
alongside the session-related constants, and use that symbol in the
futures::StreamExt pipeline so it is easy to locate and adjust independently
from crate::session::SESSION_CHECK_BATCH_SIZE.
- Around line 448-455: The two independent session-existence checks in the login
flow are run sequentially even though they query unrelated JIDs. Update the
logic around check_session_exists in src/client/sessions.rs so lid_exists and
pn_exists are evaluated concurrently rather than with back-to-back awaits,
keeping the same fallback behavior on error and preserving the existing
variables used by the login path.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI (base), Organization UI (inherited)
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 9fc6e04d-2e1f-4250-a4f7-7c2ac07eff3f
📒 Files selected for processing (1)
src/client/sessions.rs
Follow-up to oxidezap/whatsapp-rust#975, which added a concurrent batch API for recovering many expired-URL media at once.
Follow-up to oxidezap/whatsapp-rust#975, which streams each in-memory download() attempt into its own buffer so a failed host can't leave a stale tail behind a shorter successful retry.
2 participants
Summary
Parallelizes / de-serializes a set of hot paths surfaced by a concurrency audit, each cross-checked against the real WhatsApp Web client (captured JS) to confirm it matches WA Web behavior and doesn't "fix" something that must stay serial. Each item is its own commit.
The codebase was already heavily optimized (bounded parallel group-encrypt fan-out,
spawn_blockingfor prekey-gen / media-decrypt / history-decompress, batched app-state IQs, 64-wide live inbound processing), so this targets the remaining serial/blocking gaps.Changes (highest → lowest impact)
Startup / sync
Syncd/CollectionHandlerPromise.allover per-patch external mutations.set_passiveon the login pre-key upload — the upload (key-gen + store + IQ) was awaited before going active, delaying offline delivery on a fresh pairing. Now detached (with RotateKeyJob) so it no longer gates going active. Matches WA Web, which registers it as a passive task (PassiveTaskManager.registerPassiveTask("KeyUpload", …)). The upload and the signed-pre-key rotation share one detached task, ordered upload→rotate: both re-declare the signed pre-key, so overlapping them could let the upload's stale bundled key revert a freshly rotated one (see follow-ups).chunk_order), so they now ingest behind a small semaphore (cap 2, memory-conscious). WA Web caps history-sync chunks similarly (histSyncChunk=3).Media
download()— each attempt streams (CDN read + AES/HMAC decrypt interleaved in one blocking pass) into a fresh in-memory buffer, instead of fetch-whole-ciphertext-then-decrypt; overlaps network with CPU and ~halves peak memory. A fresh buffer per attempt avoids a stale-tail on host-failover (a failed longer host can't leave bytes behind a shorter successful retry).MediaReupload::request_many— new API that recovers many expired-URL media concurrently (bounded), so bulk recovery is ~one timeout instead of N×30s. Duplicatemsg_ids in a batch are rejected (past the first occurrence): themediaretrywaiter filters on id alone andresolve_waiterswakes every match, so two same-id waiters would cross-resolve — and serializing them isn't enough (a timed-out waiter lingers innode_waitersand its late notification could still resolve a same-id retry). A message id is unique per message, so a duplicate is a caller mistake.Send / device resolution
Low / micro
has_sessionconcurrently inensure_sessions(cold-cache multi-recipient setup).is_on_whatsappPN and LID queries concurrently (futures::join!).Evaluated and deliberately skipped
send_nodeserializes on the socket write lock, so parallelizing the loop only queues on that lock; the flush is already spawned off-path and aggregation leaves few stanzas.Streaming vs. batching
The one clear streaming win — buffered media download → single streaming decrypt pass — is the
media: stream buffered download()change. History-sync already streams its download and lazily decodes for consumers; the remaining "batch" points (Noise frame decode, single-query usync) are protocol-mandated and match WA Web.Runtime-agnostic / wasm
All concurrency uses runtime-agnostic primitives —
futures::join!/futures::stream::…buffer_unordered(nottokio::join!),async_lock::Semaphore, and the existingRuntime::spawnabstraction — so the lib stays wasm-buildable (--no-default-features, no direct tokio in non-test code).Review follow-ups (applied)
receive.rsnow has no functional diff.download_media_with_retry) instead of a shared writer, so a failed longer host can't leave a stale tail behind a shorter retry; kept the streaming decrypt. Added tests for the intra-attempt host-failover path (generic error → next host succeeds; all-hosts-fail →last_errpropagates) alongside the existing auth-refresh test.request_manyrejects duplicatemsg_ids (see above) so overlapping same-id waiters can't cross-resolve, closing the late-response-after-timeout window that plain serialization would leave open.download()useexecute_streaming, whose reader was unbounded (unlike bufferedexecute, capped atmax_body_bytes= WA's 2 GiB max file size). Bounded the streaming reader to the samemax_body_bytesso a CDN streaming past the declared length hits EOF (→ MAC/SHA rejection) instead of growing theVecto OOM; added a cap test.set_passivestays un-gated while rotation reads the upload's persisted snapshot.get_tc_token→ check → unconditionalstore_received_tc_token. 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. Serialized the read-check-write on a per-clienttc_token_lock(not a CAS in the shared store, which the privacy-notification path also uses).join!(nottry_join!):try_join!would drop the sibling IQ future mid-flight and leak itsresponse_waitersentry (no cancellation cleanup insend_and_wait_iq), suppressing keepalives.Declined (with reasons)
try_join!after adding waiter Drop-cleanup insend_and_wait_iq— the request-layer fix is the right long-term shape, butresponse_waitersis behind an async mutex, so cancellation-safe cleanup means converting it to a sync-lockable structure — a cross-cutting change to a hot control-plane path, out of scope for a perf PR.join!is correct today and the only cost is the rare mixed-PN+LID error path (both IQs share one 75s timeout). Left as a follow-up.Promise.allpre-download; the peak is bounded by one sync batch, and draining/streaming would re-serialize the fetch it parallelizes. Pre-existing design.receive.rs(now identical to main), so not introduced here. The two are also in tension (releasing the lock to avoid re-entrancy is exactly what broke durability); resolving it is a separatereceive.rsredesign, not part of this PR.major_sync_task_senderchannel carries onlyHistorySynctasks: nothing anywhere constructs/enqueuesMajorSyncTask::AppStateSync(app-state sync runs via its own direct path,fetch_app_state_with_retry). So no app-state task ever flows through this loop to be delayed; blocking onacquirewhen both slots are full only backpressures further history intake, which is the intended memory bound. The loop'sHistorySync-vs-elseshape is what read as if app-state flowed here — clarified with a comment (and theelsemarked defensive) so it stops being re-flagged. Even hypothetically it wouldn't be a regression (the pre-PR worker was fully serial). No code-behavior change.Testing
cargo fmt --all,cargo clippy -p whatsapp-rust --all-features --tests,cargo test -p whatsapp-rust(incl. themessage::,retry::, and media download suites);cargo test -p whatsapp-rust-ureq-http-client(body-cap tests).