perf: implement streaming decompression for history sync processing - #672
Conversation
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Repository UI Review profile: ASSERTIVE Plan: Pro Plus Run ID: 📒 Files selected for processing (2)
📝 WalkthroughSummary by CodeRabbit
WalkthroughAdds InflateReader for incremental zlib decompression and a streaming extraction path in process_history_sync (used when retain_blob == false) that parses top-level protobuf fields and messages on-the-fly, replacing eager full decompression + prost conversation decoding for the streaming case. ChangesStreaming Zlib Extraction for History Sync
Sequence Diagram(s)sequenceDiagram
participant Caller
participant process_history_sync
participant process_history_sync_streaming
participant InflateReader
participant ProtobufParser
Caller->>process_history_sync: call(compressed_blob, retain_blob=false)
process_history_sync->>process_history_sync_streaming: delegate
process_history_sync_streaming->>InflateReader: new(compressed, max)
loop while not done
process_history_sync_streaming->>InflateReader: ensure(need)
InflateReader->>InflateReader: pump() decompress into buffer
process_history_sync_streaming->>ProtobufParser: parse available bytes
ProtobufParser-->>process_history_sync_streaming: (parsed_fields, parsed_len)
process_history_sync_streaming->>InflateReader: consume(parsed_len)
end
process_history_sync_streaming-->>process_history_sync: HistorySyncExtraction (decompressed_bytes=None)
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related PRs
Alright — this must work reliably. Pay close attention to InflateReader's pump no-progress detection, max-size enforcement, and the manual field walker’s boundary handling. 🚥 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
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 `@wacore/src/history_sync.rs`:
- Around line 222-229: The match arm that tries to set result.own_pushname
contains a redundant own_user.is_some() check; remove that redundant boolean
guard and rely solely on the let-chain pattern (let Some(own) = own_user && let
Some(name) = extract_own_pushname(value, own)) so the binding happens once and
then assign result.own_pushname = Some(name) as before; update the arm that
currently checks "7 => { ... }" to use only the let-chain with
extract_own_pushname and drop own_user.is_some().
🪄 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: 9ca4508e-e883-4cb9-a40c-3fdd55005587
📒 Files selected for processing (2)
wacore/binary/src/zlib_pool.rswacore/src/history_sync.rs
Benchmark Results67 unchanged benchmark(s)
|
|
@claude review |
|
Claude finished @jlucaso1's task in 3m 6s —— View job Code Review
Overall this is a well-scoped, well-tested optimization. The streaming peak/total numbers (≈94% peak, ≈43% total) are credible because the changes target exactly the dominant cost centers: the full-blob materialization, the per- 1.
|
…g pushname arm The `let Some(own) = own_user` let-chain already short-circuits when own_user is None, so the leading own_user.is_some() check was redundant.
- InflateReader::pump now errors on a stalled stream (no output produced, no input consumed, input remaining) instead of looping forever, mirroring the one-shot decompress_zlib_pooled no-progress guard. - process_history_sync_streaming errors on a truncated FIXED32/FIXED64 tail for parity with the LENGTH_DELIMITED branch (was a silent clean break). - extract_conversation_fields skips a conversation whose id is not valid UTF-8 rather than attaching its secrets to an empty chat id. - Drop the always-true `if retain_blob` ternary on the full path (the !retain_blob case returns earlier into the streaming variant).
1 participant
What
Cuts the memory cost of history-sync decode along two axes: peak (a new streaming decompress-and-parse path) and total allocations (the per-message and per-secret-record work). All in
wacore::history_syncplus the pooled decompressor inwacore_binary::zlib_pool.Streaming decompression + incremental parse (peak). A new
InflateReaderdecompresses zlib into a bounded 64 KB window and keeps an accumulation buffer that drops its consumed prefix, so a caller can read length-delimited records as they appear.process_history_sync_streamingwalks the top-levelHistorySyncfields (conversations, pushnames, nctSalt) as their bytes arrive, extracts from each conversation, and discards it before pulling the next.process_history_syncnow branches: when the caller does not need the full decompressed blob (retain_blob == false), it streams; when a consumer wants it (Event::HistorySync), the existing whole-blob path runs unchanged. Peak shifts from the entire decompressed blob to roughly the largest single conversation.Per-message partial decode (allocations).
extract_conversation_fieldsnow walks the conversation protobuf directly and prost-decodes eachHistorySyncMsgone at a time, extracting its secret record and dropping it, instead of decoding the conversation into aVec<HistorySyncMsgInternalFields>that heap-allocated one element per message. The forwarded / poll / bot detection stays in prost via the typedMessageInternalFields, so the security-relevant logic is untouched; only the outer wrapper is read by hand.Message-secret accumulator pre-sizing (allocations). The records vector is sized once from a single allocation-free count pass (
count_history_sync_messages) and records are pushed straight into it, with no per-conversation tempVecand noextendcopy. This removes the one-element-at-a-time growth (grow_one) that dominated the decode's allocations.zlib output buffer sizing. The pooled decompressor's output buffer is now pre-sized from the compressed length (
2x, bounded by the hard cap) instead of a fixed 64 KB clamp, which had forced roughlylog2(size / 64 KB)doubling reallocations for every multi-MB chunk.Why
The history-sync core only extracts message secrets, tctokens, pushname and nctSalt during the protobuf walk; it never needs the fully materialized proto, nor every message in memory at once. Holding the whole decompressed blob, plus one heap box per message, plus a doubling accumulator, was pure cost with no decode benefit. Only an external
Event::HistorySyncconsumer actually needs the blob, and on that path nothing changes.Results
Measured with
dhaton synthetic history-sync blobs (varied per-message content, so they compress at a realistic ratio rather than collapsing to near-zero).grow_onechurn from the message vector and the secret-record accumulator goes to zero; the 60k per-message boxes go to zero.Compatibility
InflateReaderis new and additive.process_history_sync's signature is unchanged; the streaming path is selected by the existingretain_blobargument and returns the same secrets / tctokens / pushname / nctSalt as the full path (covered by a parity test).Event::HistorySyncconsumer (the caller passesretain_blob = has_handlers()). When a consumer is present, the blob is still produced via the full path, so peak there is unchanged. Broadening the gate (per-event-type interest, so a client with only unrelated handlers can still stream) is a possible follow-up.Tests
New coverage:
InflateReaderunits: round-trip read back in tiny odd-sized steps across chunk boundaries, anensure()larger than the 64 KB window, the decompressed-size cap, and one-shot vs streaming equivalence.Validation:
cargo fmt --allcargo clippy -p wacore -p wacore-binary --lib -- -D warningscleancargo test -p wacore -p wacore-binary(history-sync + zlib suites green)