perf(history-sync)!: store the compressed payload and expose a streaming reader - #853
Conversation
…ing reader - Event::HistorySync now carries the compressed bytes (~10x smaller), so queued events cost O(compressed) instead of O(decompressed) - new public HistorySyncStream: conversations one at a time with bounded memory, lenient per-conversation decode, fail-loud remainder() - the internal extractor and the stream share one wire walk (FieldWalker); the duplicated full-decompress parse path moved to cfg(test) as the parity oracle - LazyHistorySync loses the Mutex take-dance: get()/decompress()/stream() all keep working after each other, Clone is a refcount bump - fixes a latent decompress_zlib_pooled clamp panic for caps below 4096 bytes, exposed by exact-size inflate caps
📝 WalkthroughSummary by CodeRabbitRelease Notes
WalkthroughSingle-pass streaming history-sync extraction now inflates incrementally, reports exact decompressed_size, optionally retains compressed_bytes, and exposes LazyHistorySync as a compressed payload with on-demand decompress and streaming iteration. Prost build boxed nested message fields; tests and benches updated. ChangesStreaming Refactor & Boxed-proto Migration
Sequence Diagram(s)sequenceDiagram
participant Client as Sync Consumer
participant FieldWalker
participant InflateReader
participant Stream as HistorySyncStream
participant Remainder as Decode Remainder
Client->>FieldWalker: pass compressed data + cap
FieldWalker->>InflateReader: request next inflated window
InflateReader-->>FieldWalker: inflated window slice (total_out, stream_ended)
FieldWalker->>FieldWalker: parse top-level field tag/wire-type
alt conversation field
FieldWalker->>Stream: yield borrowed conversation bytes
Stream-->>Client: conversation payload
else non-conversation field
FieldWalker->>FieldWalker: buffer raw wire bytes
end
Client->>Stream: remainder()
Stream->>Remainder: decode buffered top-level fields
Remainder-->>Client: full HistorySync proto
Estimated code review effort🎯 4 (Complex) | ⏱️ ~75 minutes Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ 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: 532527b8c7
ℹ️ 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".
| .map_err(HistorySyncError::DecompressionError)? | ||
| { | ||
| break; | ||
| return Ok(None); |
There was a problem hiding this comment.
When the compressed payload is cut off after the last protobuf field but before the zlib stream terminator/checksum, this EOF path accepts the blob as successfully parsed because InflateReader treats input exhaustion as EOF. That was already risky for the non-retained streaming extractor, but this commit now routes retain_blob == true through the same walker, so clients with HistorySync handlers can dispatch a LazyHistorySync containing truncated compressed bytes; later get()/decompress() will fail even though the sync was logged and internal side effects were applied. The retained path should still force/validate zlib StreamEnd before returning success.
Useful? React with 👍 / 👎.
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
Merging this PR will not alter performance
|
| Mode | Benchmark | BASE |
HEAD |
Efficiency | |
|---|---|---|---|---|---|
| ⚡ | Memory | bench_process_history_sync |
5.8 MB | 2.5 MB | ×2.3 |
| 🆕 | Memory | bench_history_sync_stream_drain |
N/A | 542 KB | N/A |
| 🆕 | Simulation | bench_history_sync_stream_drain |
N/A | 85.2 ms | N/A |
| 👁 | Memory | bench_group_recv |
2.8 KB | 5.2 KB | -45.71% |
| 👁 | Simulation | bench_unpack_uncompressed |
183.3 ns | 241.7 ns | -24.14% |
Tip
Curious why this is faster? Comment @codspeedbot explain why this is faster on this PR, or directly use the CodSpeed MCP with your agent.
Comparing perf/history-sync-compressed-streaming (1623c5f) with main (9e8fca9)
![cubic-dev-ai[bot]](https://avatars.githubusercontent.com/in/1082092?s=60&v=4){kind=small}
There was a problem hiding this comment.
1 issue found across 5 files
Reply with feedback, questions, or to request a fix.
Fix all with cubic | Re-trigger cubic
InflateReader treated input exhaustion as clean EOF, so a blob cut exactly between protobuf fields parsed successfully and (with retain_blob) dispatched an event whose get()/decompress() would fail later. Track zlib StreamEnd explicitly and have the walker require it at EOF, matching the strictness the old full-decompress retained path had. Raised by both review bots on #853.
The stack chunk + extend_from_slice pair copied every decompressed byte a second time, ~10% of a history-sync extraction in the CodSpeed instruction profile. decompress_vec writes into the window's spare capacity instead (same idiom decompress_zlib_pooled already uses).
…decode path CodSpeed attributed ~94% of a full history-sync conversation decode to memcpy: prost's push(default) + Vec doubling moved HistorySyncMsg elements of 15,680 bytes each (WebMessageInfo inline at 15,664 with wa::Message inline at 6,504). Boxing HistorySyncMsg.message and WebMessageInfo.message collapses the element to 24 bytes; locally the stream drain drops ~25% in wall time, with a far larger instruction-count win expected. Breaking: both fields are now Option<Box<...>> (construction sites wrap with Box::new; reads auto-deref).
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 (3)
wacore/src/history_sync.rs (2)
1633-1647:⚠️ Potential issue | 🟠 Major | ⚡ Quick winDon’t emit a tc-token candidate without a conversation id.
Lines 1639-1647 only guard on
tc_token.is_empty(). If a malformed conversation carriesTC_TOKENand timestamps but omitsID, this still returnsTcTokenCandidate { id: "" }, which can leak bad data into downstream token state. Gate the return on!chat_id.is_empty()the same way message-secret extraction already does.Suggested fix
- if tc_token.is_empty() { + if chat_id.is_empty() || tc_token.is_empty() { return None; } Some(TcTokenCandidate { id: chat_id.to_string(), tc_token: tc_token.to_vec(),🤖 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 `@wacore/src/history_sync.rs` around lines 1633 - 1647, The current function can return a TcTokenCandidate with an empty conversation id; update the return gating so we only construct and return TcTokenCandidate when chat_id is non-empty and tc_token is non-empty (i.e., add a check like !chat_id.is_empty() alongside the existing tc_token.is_empty() guard). Locate the block around parse_jid_fast(...) and the return Some(TcTokenCandidate { ... }) and ensure you mirror the message-secret extraction pattern by returning None if chat_id is empty before creating TcTokenCandidate(id: chat_id.to_string(), ...).
68-75:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftRestore the exact inflate cap on the first extraction pass.
Line 73 hard-codes
MAX_DECOMPRESSED, so this pass can still inflate and process up to 64 MiB even when the producer already reported a much smallerdecompressed_size. After removing_compressed_size_hintfromprocess_history_sync, there is no path left to thread that tighter bound intoFieldWalker::new(), so the first pass lost the zip-bomb / over-allocation protection thatLazyHistorySync::decompress()applies later.Based on PR objectives and the downstream
LazyHistorySynccontract, the producer-reported decompressed size is supposed to be the cap, not just a metric.🤖 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 `@wacore/src/history_sync.rs` around lines 68 - 75, The initial extraction pass currently uses the hard-coded MAX_DECOMPRESSED in process_history_sync when calling process_history_sync_streaming, which loses the producer-reported decompressed_size cap; change the call so the producer-reported decompressed_size (when available) is threaded into process_history_sync_streaming and ultimately into FieldWalker::new (or reintroduce a decompressed_size_hint parameter to process_history_sync) so the first pass enforces the same cap LazyHistorySync::decompress() uses instead of MAX_DECOMPRESSED; ensure the compressed retain_blob path and returned HistorySyncResult still work with the new parameter.src/history_sync.rs (1)
144-148:⚠️ Potential issue | 🟠 Major | 🏗️ Heavy liftThis snapshots
HistorySyncinterest too early.At Lines 144-148 you decide whether to retain the blob before
process_history_syncstarts.CoreEventBussays interest is re-checked at dispatch time, but this path makesEvent::HistorySyncimpossible to materialize if a handler widens toEventKind::HistorySyncduring the parse window. On large blobs that window is real because Lines 160-171 push the work intospawn_blocking, and Lines 217-231 only dispatch whencompressed_byteswas retained up front. Either carry the compressed blob through dispatch-time interest evaluation, or explicitly narrow the bus contract forHistorySync.🤖 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/history_sync.rs` around lines 144 - 148, The code snapshots HistorySync interest too early by evaluating retain_history_blob = self.core.event_bus.has_handler_for(EventKind::HistorySync) before process_history_sync and then dropping compressed_bytes unless that pre-check passed; instead, change the logic so compressed_bytes is carried through to dispatch-time interest evaluation (or explicitly narrow the contract) — i.e., remove/avoid using the early retain_history_blob flag, pass compressed_bytes along into the spawn_blocking/dispatch path, and call self.core.event_bus.has_handler_for(EventKind::HistorySync) (or use the bus’s dispatch-time check) right before creating/dispatching Event::HistorySync in process_history_sync so a handler that registers during parsing can still cause the blob to be materialized.
🤖 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/history_sync.rs`:
- Around line 144-148: The code snapshots HistorySync interest too early by
evaluating retain_history_blob =
self.core.event_bus.has_handler_for(EventKind::HistorySync) before
process_history_sync and then dropping compressed_bytes unless that pre-check
passed; instead, change the logic so compressed_bytes is carried through to
dispatch-time interest evaluation (or explicitly narrow the contract) — i.e.,
remove/avoid using the early retain_history_blob flag, pass compressed_bytes
along into the spawn_blocking/dispatch path, and call
self.core.event_bus.has_handler_for(EventKind::HistorySync) (or use the bus’s
dispatch-time check) right before creating/dispatching Event::HistorySync in
process_history_sync so a handler that registers during parsing can still cause
the blob to be materialized.
In `@wacore/src/history_sync.rs`:
- Around line 1633-1647: The current function can return a TcTokenCandidate with
an empty conversation id; update the return gating so we only construct and
return TcTokenCandidate when chat_id is non-empty and tc_token is non-empty
(i.e., add a check like !chat_id.is_empty() alongside the existing
tc_token.is_empty() guard). Locate the block around parse_jid_fast(...) and the
return Some(TcTokenCandidate { ... }) and ensure you mirror the message-secret
extraction pattern by returning None if chat_id is empty before creating
TcTokenCandidate(id: chat_id.to_string(), ...).
- Around line 68-75: The initial extraction pass currently uses the hard-coded
MAX_DECOMPRESSED in process_history_sync when calling
process_history_sync_streaming, which loses the producer-reported
decompressed_size cap; change the call so the producer-reported
decompressed_size (when available) is threaded into
process_history_sync_streaming and ultimately into FieldWalker::new (or
reintroduce a decompressed_size_hint parameter to process_history_sync) so the
first pass enforces the same cap LazyHistorySync::decompress() uses instead of
MAX_DECOMPRESSED; ensure the compressed retain_blob path and returned
HistorySyncResult still work with the new parameter.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 3929c703-074f-4e54-961b-e1113c7b37ba
📒 Files selected for processing (7)
src/history_sync.rssrc/pdo.rswacore/benches/history_sync_benchmark.rswacore/binary/src/zlib_pool.rswacore/src/history_sync.rswacore/src/types/events.rswaproto/build.rs
…mpty-id tctoken candidates The pre-parse has_handler_for snapshot dated from when retain meant a full decompress; now that retaining is a free move of the compressed input, snapshotting early only made a handler registered during a long parse silently miss the event. The blob is always carried through and interest is checked right before dispatch. extract_conversation_fields could also return a TcTokenCandidate with an empty conversation id (the downstream JID parse dropped it, but the API should not produce it); it now mirrors the message-secret empty-id guard. A third reviewer suggestion (threading a producer decompressed-size cap into the first pass) was skipped: that size does not exist before the parse, it is what the extraction pass counts.
There was a problem hiding this comment.
Actionable comments posted: 2
🤖 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/history_sync.rs`:
- Around line 214-219: Remove the early dispatch-time gate using
self.core.event_bus.has_handler_for(EventKind::HistorySync); instead only check
for sync_result.compressed_bytes, construct the LazyHistorySync from that
compressed payload, and call self.core.event_bus.dispatch(...) so
CoreEventBus::dispatch() performs the final interest check. This eliminates the
missed-event race by deferring the handler-interest decision to
CoreEventBus::dispatch and keeps references to sync_result.compressed_bytes and
LazyHistorySync intact.
🪄 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
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 3bd99332-3ff3-4ef7-9276-4e7b2134bcdb
📒 Files selected for processing (2)
src/history_sync.rswacore/src/history_sync.rs
There was a problem hiding this comment.
1 issue found across 2 files (changes from recent commits).
Tip: Review your code locally with the cubic CLI to iterate faster.
Fix all with cubic | Re-trigger cubic
The dispatch-time has_handler_for pre-check read a different bus snapshot than dispatch() itself, leaving a check-to-dispatch window where a freshly registered handler lost the event — the same race class the previous commit removed, just narrower. dispatch() already evaluates interest against a single snapshot and skips materializing the Arc when nobody listens, and building the event is only a Bytes refcount move, so the pre-check bought nothing.
Second boxing wave, found via a full size_of probe of every inline field: WebMessageInfo.statusMentionMessageInfo carries a wa::Message inline and was most of WebMessageInfo's remaining bulk; messageContextInfo was the only large inline left in wa::Message. Drops the per-message default-construct copy from ~15.7 KB to ~6.5 KB (WebMessageInfo 9168 -> 2672 B, Message 6504 -> 3784 B). Wire and serde shapes are unchanged; constructors wrap with Box::new, reads auto-deref.
|
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.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
wacore/src/messages.rs (1)
377-386:⚠️ Potential issue | 🟡 MinorFix
unwrap_device_sentto use a let-chain (no nestedif let)
wacore/src/messages.rsstill uses nestedif letinsideunwrap_device_sent(thedevice_sent_message/dsm.messageunwrap), which violates the let-chain guideline. Convert it to a single let-chain while preserving the current behavior (return the inner message whendsm.messageexists; otherwise keep the wrapper).Proposed refactor
pub fn unwrap_device_sent(mut msg: wa::Message) -> wa::Message { - if let Some(mut dsm) = msg.device_sent_message.take() { - if let Some(mut inner) = dsm.message.take() { - inner.message_context_info = crate::proto_helpers::merge_dsm_context( - inner.message_context_info.take(), - msg.message_context_info.as_deref(), - ); - return *inner; - } - msg.device_sent_message = Some(dsm); + if let Some(dsm) = msg.device_sent_message.as_mut() + && let Some(mut inner) = dsm.message.take() + { + inner.message_context_info = crate::proto_helpers::merge_dsm_context( + inner.message_context_info.take(), + msg.message_context_info.as_deref(), + ); + return *inner; } msg }🤖 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 `@wacore/src/messages.rs` around lines 377 - 386, Refactor unwrap_device_sent to remove nested if-let by first taking msg.device_sent_message into a local variable (e.g., let mut dsm_opt = msg.device_sent_message.take()), then use a single let-chain: if let Some(mut dsm) = dsm_opt && let Some(mut inner) = dsm.message.take() { update inner.message_context_info via crate::proto_helpers::merge_dsm_context and return *inner; } else if let Some(dsm) = dsm_opt { restore msg.device_sent_message = Some(dsm); } — this preserves the existing behavior around msg.device_sent_message, dsm.message, merge_dsm_context, and inner.message_context_info while complying with the let-chain guideline.Source: Coding guidelines
🤖 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 `@wacore/src/messages.rs`:
- Around line 377-386: Refactor unwrap_device_sent to remove nested if-let by
first taking msg.device_sent_message into a local variable (e.g., let mut
dsm_opt = msg.device_sent_message.take()), then use a single let-chain: if let
Some(mut dsm) = dsm_opt && let Some(mut inner) = dsm.message.take() { update
inner.message_context_info via crate::proto_helpers::merge_dsm_context and
return *inner; } else if let Some(dsm) = dsm_opt { restore
msg.device_sent_message = Some(dsm); } — this preserves the existing behavior
around msg.device_sent_message, dsm.message, merge_dsm_context, and
inner.message_context_info while complying with the let-chain guideline.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: 0a604161-9107-4b87-8342-803e2e01cd68
📒 Files selected for processing (14)
src/client/messaging.rssrc/features/comments.rssrc/features/events.rssrc/features/polls.rssrc/history_sync.rssrc/message/tests.rssrc/send.rswacore/benches/history_sync_benchmark.rswacore/src/history_sync.rswacore/src/messages.rswacore/src/msg_secret.rswacore/src/proto_helpers.rswacore/src/reporting_token.rswaproto/build.rs
…xtractor 10k seeded byte flips/truncations of a valid blob must never panic on either consumption path, only clean errors or lenient skips. Added during an adversarial review pass of the PR.
|
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.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
wacore/src/messages.rs (1)
377-383:⚠️ Potential issue | 🟡 MinorCollapse the nested
if letguards inunwrap_device_sentinto a single let-chain.
unwrap_device_sentstill uses nestedif let(device_sent_message.take()thendsm.message.take()), so it violates the repo’scollapsible_if/let-chains rule. Convert it to a let-chain while keeping the current behavior that re-attachesdsmback ontomsg.device_sent_messagewhendsm.messageisNone.🤖 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 `@wacore/src/messages.rs` around lines 377 - 383, Replace the nested if-let in unwrap_device_sent by first doing let mut dsm_opt = msg.device_sent_message.take(); then use a let-chain: if let Some(mut dsm) = dsm_opt && let Some(mut inner) = dsm.message.take() { update inner.message_context_info via crate::proto_helpers::merge_dsm_context and return *inner; } else if let Some(dsm) = dsm_opt { re-attach it with msg.device_sent_message = Some(dsm); } — this preserves behavior while collapsing the nested guards (references: unwrap_device_sent, msg.device_sent_message, dsm, dsm.message, inner, crate::proto_helpers::merge_dsm_context).Source: Coding guidelines
🤖 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 `@wacore/src/messages.rs`:
- Around line 377-383: Replace the nested if-let in unwrap_device_sent by first
doing let mut dsm_opt = msg.device_sent_message.take(); then use a let-chain: if
let Some(mut dsm) = dsm_opt && let Some(mut inner) = dsm.message.take() { update
inner.message_context_info via crate::proto_helpers::merge_dsm_context and
return *inner; } else if let Some(dsm) = dsm_opt { re-attach it with
msg.device_sent_message = Some(dsm); } — this preserves behavior while
collapsing the nested guards (references: unwrap_device_sent,
msg.device_sent_message, dsm, dsm.message, inner,
crate::proto_helpers::merge_dsm_context).
ℹ️ Review info
⚙️ Run configuration
Configuration used: Repository UI
Review profile: ASSERTIVE
Plan: Pro Plus
Run ID: a0cda85f-3e34-4ea9-81f1-adc9670e7727
📒 Files selected for processing (14)
src/client/messaging.rssrc/features/comments.rssrc/features/events.rssrc/features/polls.rssrc/history_sync.rssrc/message/tests.rssrc/send.rswacore/benches/history_sync_benchmark.rswacore/src/history_sync.rswacore/src/messages.rswacore/src/msg_secret.rswacore/src/proto_helpers.rswacore/src/reporting_token.rswaproto/build.rs
1 participant
Why
Event::HistorySynccarried the fully decompressed blob (a typical InitialBootstrap chunk is 5-20 MB inflated, ~10x the compressed size), so any consumer with a HistorySync handler paid O(decompressed) retained memory per queued event, and incremental consumption required hand-rolling a protobuf wire walk. Internally we also kept two copies of that walk (a streaming one and a full-buffer one) guarded by a parity test.What changed
The event stores compressed bytes.
LazyHistorySyncnow holds the original zlib payload as one immutableBytesplus the exactdecompressed_sizecounted by the extraction pass. Holding or queueing the event costs O(compressed). The Mutex + take-dance + manual lazy/deep Clone are gone:Cloneis a refcount bump, andget(),decompress()andstream()all keep working after each other (previouslyraw_bytes()returnedNoneafterget()).Public streaming reader.
wacore::history_sync::HistorySyncStreamiterates a compressed blob with bounded memory (peak ≈ the largest single conversation):next_conversation_bytes()lends raw entry bytes from the inflate window,next_conversation()is the decoded convenience layer and is lenient (a corrupt entry is skipped and counted viaskipped_conversations(), not fatal), andremainder()decodes everything that is not a conversation (pushnames, mappings, nctSalt, ...) regardless of wire order. Callingremainder()before exhausting the conversations drains the tail and fails loud with the newHistorySyncError::UnreadConversationsif it would silently drop one.One wire walk. Both the internal extractor (
process_history_sync) and the public stream consume the sameFieldWalker, so the format knowledge lives in exactly one production place. The old full-decompress parse path was deleted from production and moved into the test module asreference_full_walk, keeping the differential parity oracle alive (production stream vs independent full-buffer implementation) instead of weakening it to a re-derived check.Extraction always streams.
retain_blob == truenow just hands the compressed input back inHistorySyncResult::compressed_bytes(aVectoBytesmove, no copy, no second inflate), so a bot with a catch-allon_eventhandler no longer forces the whole-blob materialization on every chunk. The unused_compressed_size_hintparameter is gone.Exact-size inflate caps. The producer-counted
decompressed_sizeis used as the inflate bound forget()/decompress()/stream(), a strictly tighter anti-bomb limit than the global 64 MB ceiling (MAX_DECOMPRESSED, now apub constfor rawHistorySyncStream::newusers).InflateReadergained atotal_out()getter to expose the count.Latent panic fixed. Exercising small exact-size caps surfaced a real bug:
decompress_zlib_pooledpre-sizes its buffer with.clamp(4096, cap), which panics whenever the cap is below 4096 bytes. It never fired before because the only caller passed 64 MB, but any small-cap call (a PushName-only chunk through the newdecompress()) would have hit it. The floor now bows to the cap, with a regression test.Cost model (documented on the new APIs)
Consumers that decode pay one extra zlib inflate per consumption pass compared to the old retained-decompressed design; in exchange every queued event is ~10x smaller. A multi-MB chunk takes tens of milliseconds to inflate (plus prost decode for
get()), so the rustdoc onLazyHistorySyncandHistorySyncStreamrecommendsspawn_blocking(cloningcompressed_bytes()into the closure) whendecompressed_size()is large, mirroring the producer's own 256 KB inline threshold.Breaking changes and migration
LazyHistorySync::new(raw, sync_type, ...)is nownew(compressed, decompressed_size, sync_type, ...).raw_bytes()was removed: usedecompress()for the inflated bytes (per call, no caching),compressed_bytes()for the stored payload, orstream()for incremental consumption.raw_size()is nowdecompressed_size().HistorySyncResult::decompressed_byteswas replaced bycompressed_bytesplusdecompressed_size;process_history_synclost its unused 4th parameter.Clone(a clone re-inflates on demand);Serializeoutput is unchanged (metadata only).Tests
44 wacore history-sync tests (stream parity vs full prost decode, field-order-shuffled blobs, conversation-less blobs, lenient decode, zero-length conversation, truncated zlib and truncated length-delimited fields, window growth for a 1 MB conversation, repeated window refills, decompressed cap enforcement, fail-loud early
remainder(), unknown wire types, exact size reporting), 12LazyHistorySynctests (everything-works-after-get, per-call decompress, cheap clone, undersized-cap fail-loud), and a new producer test asserting the dispatched event carries the original compressed payload + exact size and thatget()/stream()work end-to-end. The differential corpus and prior extraction tests pass unchanged.Validation:
cargo clippy --workspace --all-targets -- -D warningsclean, full test suites green (whatsapp-rust 984, wacore 799+103, wacore-binary),cargo bench --no-runcompiles (the bench dropped the removed parameter and gained astream_drainconsumer-side benchmark), wasm32--no-default-featuresbuild green. Expect CodSpeed movement onbench_process_history_sync: the retain path no longer does the duplicate full-buffer walk, so it should get faster.Addendum: flamegraph-driven follow-ups (same PR)
CodSpeed flamegraph analysis of the first revision surfaced two more wins, both included here:
Boxed per-message giants (
perf(waproto)!). The newstream_drainbenchmark exposed that ~94% of a full conversation decode was memcpy: prost'spush(default)plusVecdoubling movedHistorySyncMsgelements of 15,680 bytes each (WebMessageInfoinline at 15,664 bytes, withwa::Messageinline at 6,504 bytes).HistorySyncMsg.messageandWebMessageInfo.messageare now generated asOption<Box<...>>(prost_build::Config::boxed), collapsing the element to 24 bytes. Locally the stream drain improves ~25% in wall time; the instruction-count (Simulation) win should be substantially larger. Breaking for constructors (Some(x)becomesSome(Box::new(x))); reads auto-deref.Single-copy inflate (
perf(zlib)).InflateReader::pumpinflated into a 64 KB stack chunk and thenextend_from_sliced into the window, copying every decompressed byte twice (~10% of an extraction in the instruction profile). It now inflates straight into the window's spare capacity viadecompress_vec, the same idiomdecompress_zlib_pooledalready used.Also addressed from review: both bots flagged that a zlib stream truncated exactly at a protobuf field boundary passed extraction as clean EOF (the old full-decompress retained path rejected it).
InflateReadernow tracks a real zlibStreamEndand the walker requires it at EOF, with a sync-flush-without-finish regression test.