feat!: replace history sync events with lazy blob + perf optimizations - #533
Conversation
…blob Replace `Event::JoinedGroup(LazyConversation)` (dispatched N times per history sync) and dead `Event::HistorySync(HistorySync)` with a single `Event::HistorySync(Box<LazyHistorySync>)` per sync blob. LazyHistorySync wraps the full decompressed protobuf bytes with lazy decode via OnceLock. Consumers get: - Cheap metadata without decode: sync_type(), chunk_order(), progress() - Full proto access on demand: .get() -> Option<&wa::HistorySync> - Raw bytes for custom/partial decoding: .raw_bytes() Internal processing (tctokens, pushname, nct_salt) is preserved — streaming extraction via channel still runs before event dispatch. BREAKING: Event::JoinedGroup removed, Event::HistorySync payload changed, LazyConversation removed.
|
Note Reviews pausedIt looks like this branch is under active development. To avoid overwhelming you with review comments due to an influx of new commits, CodeRabbit has automatically paused this review. You can configure this behavior by changing the Use the following commands to manage reviews:
Use the checkboxes below for quick actions:
📝 WalkthroughWalkthroughSwaps per-conversation streaming/lazy decode for a single post-parse wrapper: history-sync parsing now returns decompressed bytes and tc-token candidates; tctokens are stored from parsed candidates; a single Changes
Sequence Diagram(s)sequenceDiagram
participant Task as HistorySyncTask
participant Parser as process_history_sync
participant Store as TCTokenStore
participant EventBus as EventBus
Task->>Parser: parse compressed payload (retain_blob flag)
Parser-->>Task: HistorySyncResult{tc_token_candidates, decompressed_bytes}
loop store candidates
Task->>Store: store_tc_token_candidate(candidate)
end
Task->>EventBus: dispatch Event::HistorySync(Box::new(LazyHistorySync::new(decompressed_bytes,...)))
Estimated Code Review Effort🎯 4 (Complex) | ⏱️ ~45 minutes Possibly Related PRs
Poem
🚥 Pre-merge checks | ✅ 3✅ Passed checks (3 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
🧪 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 |
Benchmark Results59 unchanged benchmark(s)
|
![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
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (1)
wacore/src/types/events.rs (1)
97-107:⚠️ Potential issue | 🟠 MajorFix non-deterministic serialization of
LazyHistorySyncto preserve metadata.Currently,
Event::HistorySyncserializes asnulluntil some caller invokesget(), then starts serializing the full protobuf. This violates deterministic serialization, dropssync_type/chunk_order/progressentirely for untouched events, and contradicts the type's documented purpose of providing "cheap metadata for filtering without decoding."♻️ Always include metadata in serialized form
impl Serialize for LazyHistorySync { fn serialize<S>(&self, serializer: S) -> Result<S::Ok, S::Error> where S: serde::Serializer, { - if let Some(Some(hs)) = self.parsed.get() { - hs.serialize(serializer) - } else { - serializer.serialize_none() - } + #[derive(Serialize)] + struct SerializableLazyHistorySync<'a> { + sync_type: i32, + chunk_order: Option<u32>, + progress: Option<u32>, + #[serde(skip_serializing_if = "Option::is_none")] + history_sync: Option<&'a wa::HistorySync>, + } + + SerializableLazyHistorySync { + sync_type: self.sync_type, + chunk_order: self.chunk_order, + progress: self.progress, + history_sync: self.get(), + } + .serialize(serializer) } }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 97 - 107, The Serialize impl for LazyHistorySync must always emit the lightweight metadata (sync_type, chunk_order, progress) instead of returning null when parsed.get() is None; update the impl for LazyHistorySync so it builds and serializes a struct/object that always includes those metadata fields and, if let Some(Some(hs)) = self.parsed.get() then includes the full parsed protobuf payload as an additional field (or serializes hs inline), otherwise omits or serializes that payload as None — this preserves deterministic serialization for Event::HistorySync while still including the full decoded data when available.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@wacore/src/types/events.rs`:
- Around line 97-107: The Serialize impl for LazyHistorySync must always emit
the lightweight metadata (sync_type, chunk_order, progress) instead of returning
null when parsed.get() is None; update the impl for LazyHistorySync so it builds
and serializes a struct/object that always includes those metadata fields and,
if let Some(Some(hs)) = self.parsed.get() then includes the full parsed protobuf
payload as an additional field (or serializes hs inline), otherwise omits or
serializes that payload as None — this preserves deterministic serialization for
Event::HistorySync while still including the full decoded data when available.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 7e7652ce-44c1-4de3-98f9-022911e887e3
📒 Files selected for processing (3)
src/history_sync.rswacore/src/history_sync.rswacore/src/types/events.rs
|
@claude review |
|
Claude finished @jlucaso1's task in 3m 18s —— View job Code Review: Replace per-conversation history sync events with single lazy blob
Verdict: Looks good overallThis is a well-motivated refactor that cleans up three problems at once: a misleadingly-named event ( What's done wellUnified code paths (
Zero-copy chain preserved (
Test coverage: 7 tests covering the key behaviors (decode, caching via pointer equality, metadata access, raw bytes, empty bytes, corrupt bytes, message preservation). Issues / suggestions1. Serialize impl drops metadata when unparsed (minor)
CodeRabbit flagged this too and I agree it's worth fixing, though it's not a blocker. If any consumer logs or persists serialized events without first calling 2. Memory trade-off worth documenting (informational)The old design dispatched N This is a reasonable trade-off (1 event vs N events, simpler consumer API, access to pushnames/global settings/etc.), but the full proto can be large for initial bootstrap syncs. The blob lives until all handlers drop the 3.
|
The Serialize impl now always emits sync_type, chunk_order, and progress fields. The decoded proto is included as "data" when available, null otherwise. Avoids returning bare null when the proto hasn't been decoded yet.
![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: 932816bfa1
ℹ️ 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".
| if self.core.event_bus.has_handlers() { | ||
| let lazy_hs = LazyHistorySync::new( | ||
| sync_result.decompressed_bytes, | ||
| notification.sync_type().into(), |
There was a problem hiding this comment.
Event::HistorySync metadata is populated with notification.sync_type().into(), but the prost-generated sync_type() accessor returns the default enum when the field is unset or contains an unknown value. That means new/forward-compat server sync types get silently rewritten (typically to InitialBootstrap) before handlers see them, so consumers branching on hs.sync_type() can take the wrong path. Prefer preserving the raw field value (notification.sync_type) and representing it as optional/raw in LazyHistorySync metadata.
Useful? React with 👍 / 👎.
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@wacore/src/types/events.rs`:
- Around line 97-108: The Serialize impl for LazyHistorySync currently calls
self.parsed.get() and may mutate shared OnceLock state, causing
handler-order-dependent outputs; update the Serialize implementation for
LazyHistorySync to avoid invoking parsed.get() (or any method that
initializes/mutates OnceLock) and only emit metadata fields (sync_type,
chunk_order, progress) or a boolean flag like has_data, making inclusion of the
full decoded data an explicit opt-in via a separate method/serializer; locate
the Serialize impl for LazyHistorySync and remove the data field serialization
(or replace with a non-mutating indicator) so serialization is deterministic and
side-effect free.
- Around line 68-74: The current get method runs wa::HistorySync::decode inline
via self.parsed.get_or_init, which can block the calling async Tokio worker;
change this so the initial full decode is performed on a blocking thread: either
replace get with an async offloaded accessor (e.g., async fn
get_offloaded(&self) -> Option<&wa::HistorySync>) that uses
tokio::task::spawn_blocking to run wa::HistorySync::decode and then stores the
result into self.parsed (or provide a separate init_offloaded() that does the
spawn_blocking decode and populates parsed before async handlers call get),
ensuring all references to get in history_sync.rs use the offloaded
initializer/accessor and that parsed and wa::HistorySync::decode are the unique
symbols updated.
- Around line 76-79: The getter raw_bytes currently returns a slice (&[u8])
which drops the cheap ref-counted Bytes ownership and forces copies; update the
events type so the backing field stores bytes::Bytes (or keep it if already
Bytes) and change pub fn raw_bytes(&self) -> &[u8] to a public API that returns
Bytes (i.e., pub fn raw_bytes(&self) -> Bytes) so callers (like
LazyHistorySync::new and HistorySyncResult.decompressed_bytes consumers) can
retain/forward the blob zero-copy via Bytes::clone; ensure the method clones the
internal Bytes for the return and update any callers accordingly to accept
Bytes.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: d76bf170-a92f-4b22-a4e2-60ac77c6a6db
📒 Files selected for processing (1)
wacore/src/types/events.rs
| /// Access the raw decompressed protobuf bytes for custom/partial decoding. | ||
| pub fn raw_bytes(&self) -> &[u8] { | ||
| &self.raw_bytes | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion | 🟠 Major
Keep Bytes on the public raw-bytes API.
LazyHistorySync::new() is fed a Bytes from HistorySyncResult.decompressed_bytes (src/history_sync.rs:248-253), but this getter narrows it to &[u8]. Callers that need to retain or forward the blob now lose the cheap ref-counted handoff and have to allocate a copy.
♻️ One zero-copy option
- pub fn raw_bytes(&self) -> &[u8] {
- &self.raw_bytes
+ pub fn raw_bytes(&self) -> Bytes {
+ self.raw_bytes.clone()
}🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@wacore/src/types/events.rs` around lines 76 - 79, The getter raw_bytes
currently returns a slice (&[u8]) which drops the cheap ref-counted Bytes
ownership and forces copies; update the events type so the backing field stores
bytes::Bytes (or keep it if already Bytes) and change pub fn raw_bytes(&self) ->
&[u8] to a public API that returns Bytes (i.e., pub fn raw_bytes(&self) ->
Bytes) so callers (like LazyHistorySync::new and
HistorySyncResult.decompressed_bytes consumers) can retain/forward the blob
zero-copy via Bytes::clone; ensure the method clones the internal Bytes for the
return and update any callers accordingly to accept Bytes.
There was a problem hiding this comment.
♻️ Duplicate comments (3)
wacore/src/types/events.rs (3)
82-83:⚠️ Potential issue | 🟠 MajorKeep
Byteson the public raw-bytes API.Returning
&[u8]drops the cheap ref-counted handoff thatBytesgives you, so any caller that needs to retain or forward the blob now has to copy it. That defeats the zero-copy path for large history-sync payloads.♻️ Minimal API fix
- pub fn raw_bytes(&self) -> &[u8] { - &self.raw_bytes + pub fn raw_bytes(&self) -> Bytes { + self.raw_bytes.clone() }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 82 - 83, The public accessor raw_bytes currently returns a &[u8], losing the cheap ref-counted semantics of Bytes; change the API to return Bytes instead (keep the field name raw_bytes) so callers can retain/forward without copying — update pub fn raw_bytes(&self) -> Bytes to return a cloned handle of the internal Bytes (i.e., return the ref-counted Bytes, not a slice) and adjust any call sites/tests expecting &[u8] accordingly.
75-78:⚠️ Potential issue | 🟠 MajorDon't run the first full protobuf decode inline.
wa::HistorySync::decodeon these blobs can be expensive enough to stall whichever thread hitsget()first, including Tokio workers. Please move the initial decode behind aspawn_blocking-backed initializer/accessor and keepget()as a cached read.As per coding guidelines, "All I/O uses Tokio; wrap blocking I/O (
ureq) and heavy CPU work intokio::task::spawn_blocking".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 75 - 78, The get() method currently does a synchronous, potentially expensive wa::HistorySync::decode on first access via self.parsed.get_or_init; change this so the initial decode is performed inside a tokio::task::spawn_blocking and stored in the cache asynchronously, while get() remains a cheap cached read. Concretely: replace the inline decode in the initializer for parsed with an async-safe path that kicks off spawn_blocking to decode self.raw_bytes into wa::HistorySync once (or await a stored JoinHandle/future), then store the decoded value into parsed (or swap in when ready); ensure subsequent calls to get() return the cached Option<&wa::HistorySync> without performing blocking work. Use the symbols get, parsed, raw_bytes, and wa::HistorySync to locate the change.
102-113:⚠️ Potential issue | 🟠 MajorMake event serialization deterministic.
Because the same
LazyHistorySyncinstance is shared across handlers, this emits eitherdata: nullor the full decoded proto depending on whether some earlier consumer already calledget(). That makes JSON output handler-order dependent and can unexpectedly dump full sync contents into logs or telemetry. KeepSerializemetadata-only, and make full-payload serialization explicit.♻️ One deterministic option
- let mut s = serializer.serialize_struct("LazyHistorySync", 4)?; + let mut s = serializer.serialize_struct("LazyHistorySync", 3)?; s.serialize_field("sync_type", &self.sync_type)?; s.serialize_field("chunk_order", &self.chunk_order)?; s.serialize_field("progress", &self.progress)?; - s.serialize_field("data", &self.parsed.get().and_then(|o| o.as_ref()))?; s.end()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 102 - 113, The current Serialize impl for LazyHistorySync reads self.parsed.get() and conditionally emits the full decoded payload, making JSON output handler-order dependent; change the Serialize implementation for LazyHistorySync to only emit metadata (serialize "sync_type", "chunk_order", "progress") and serialize the "data" field as None (or omit it) instead of calling self.parsed.get(); keep the existing field names ("sync_type", "chunk_order", "progress", "data") so consumers still see metadata, and if full payload serialization is needed add an explicit method (e.g., LazyHistorySync::serialize_full or a separate serializer) that intentionally accesses self.parsed.get() to return the decoded proto.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Duplicate comments:
In `@wacore/src/types/events.rs`:
- Around line 82-83: The public accessor raw_bytes currently returns a &[u8],
losing the cheap ref-counted semantics of Bytes; change the API to return Bytes
instead (keep the field name raw_bytes) so callers can retain/forward without
copying — update pub fn raw_bytes(&self) -> Bytes to return a cloned handle of
the internal Bytes (i.e., return the ref-counted Bytes, not a slice) and adjust
any call sites/tests expecting &[u8] accordingly.
- Around line 75-78: The get() method currently does a synchronous, potentially
expensive wa::HistorySync::decode on first access via self.parsed.get_or_init;
change this so the initial decode is performed inside a
tokio::task::spawn_blocking and stored in the cache asynchronously, while get()
remains a cheap cached read. Concretely: replace the inline decode in the
initializer for parsed with an async-safe path that kicks off spawn_blocking to
decode self.raw_bytes into wa::HistorySync once (or await a stored
JoinHandle/future), then store the decoded value into parsed (or swap in when
ready); ensure subsequent calls to get() return the cached
Option<&wa::HistorySync> without performing blocking work. Use the symbols get,
parsed, raw_bytes, and wa::HistorySync to locate the change.
- Around line 102-113: The current Serialize impl for LazyHistorySync reads
self.parsed.get() and conditionally emits the full decoded payload, making JSON
output handler-order dependent; change the Serialize implementation for
LazyHistorySync to only emit metadata (serialize "sync_type", "chunk_order",
"progress") and serialize the "data" field as None (or omit it) instead of
calling self.parsed.get(); keep the existing field names ("sync_type",
"chunk_order", "progress", "data") so consumers still see metadata, and if full
payload serialization is needed add an explicit method (e.g.,
LazyHistorySync::serialize_full or a separate serializer) that intentionally
accesses self.parsed.get() to return the decoded proto.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: c96e1dda-482c-4063-894c-b963734a15aa
📒 Files selected for processing (1)
wacore/src/types/events.rs
- Box OnceLock cached value to avoid inline HistorySync-sized hole - Safe Clone impl that shares Bytes cheaply without deep-copying the decoded proto - Add into_raw_bytes() for zero-copy Bytes ownership transfer - Remove async_channel from history sync — collect tctoken candidates directly in the streaming loop and process after completion - Conditional blob retention: only keep decompressed Bytes when event listeners exist (saves memory for headless bots) - Raise decompression estimate cap from 8MB to 64MB to avoid unnecessary reallocs on large InitialBootstrap blobs - Skip spawn_blocking for small blobs (<256KB) — decode inline to avoid thread hop overhead for PushName/Recent syncs - Manual pushname parser that checks id before allocating, avoiding string allocations for 99% of non-matching entries - Use prost partial decode for tctokens (DRY, readable) with parse_jid_fast early-exit for groups/newsletters/bots - Add conversations_raw() iterator for zero-copy per-conversation access without full proto decode
|
@claude review things carefully: performance, unnecessary allocations, what can be dry, also try to find dead code |
jlucaso1
changed the title
|
Claude finished @jlucaso1's task in 3m 8s —— View job Deep Review: Performance, Allocations, DRY & Dead Code
Verdict: Good refactor. A few concrete findings below.1. Dead code:
|
| # | Severity | Type | Description |
|---|---|---|---|
| 1 | Medium | Dead code | into_raw_bytes, conversations_raw, ConversationIterator — 50+ lines with zero callers and no tests |
| 2 | Low | Allocation | tc_token.clone() avoidable by taking ownership in the iteration loop |
| 3 | Low | Correctness | Serialize data field is non-deterministic (handler-order dependent) |
| 4 | Informational | Config | Decompression cap raised from 8 MB to 64 MB — worth a comment if intentional |
| 5 | Informational | Robustness | extract_own_pushname assumes canonical field ordering |
| 6 | Informational | Allocation | Prost partial decode allocates per-conversation even for filtered groups |
| 7 | Informational | Pre-existing | message_id.clone() avoidable by reordering |
Items 1-3 are worth addressing. The rest is informational. Overall this is a clean simplification — the removal of the channel-based streaming and per-conversation dispatch is a net win for both code clarity and maintenance.
There was a problem hiding this comment.
Actionable comments posted: 1
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
src/history_sync.rs (1)
256-277:⚠️ Potential issue | 🟠 MajorMake the tc-token merge atomic.
This is still a read/modify/write race: a newer post-send
put_tc_tokencan land afterget_tc_token()and then be overwritten by this history-sync write, so the newersender_timestampis not actually protected. The max-merge needs to happen inside one backend operation or transactional persistence command.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/history_sync.rs` around lines 256 - 277, The current read/modify/write in history_sync.rs using backend.get_tc_token and backend.put_tc_token allows a race that can overwrite a newer sender_timestamp; change this to perform the max-merge inside a single backend operation or transaction. Add or use an atomic backend method (e.g., backend.merge_tc_token or a transactional put method) that takes token_key and the incoming candidate/tc values and computes the merged TcTokenEntry (max of existing.sender_timestamp and candidate.tc_token_sender_timestamp, and appropriate token_timestamp logic) inside the backend implementation, then persist it in one step; update the call site in history_sync.rs to call that atomic method instead of calling get_tc_token followed by put_tc_token so the sender_timestamp merge is guaranteed atomic.wacore/src/history_sync.rs (1)
99-107: 🛠️ Refactor suggestion | 🟠 MajorDrop the
unwrap()from the pushname scan.This branch is already conditioned on
own_user, so theunwrap()is avoidable and violates the repo rule for Rust code. A let-chain keeps the arm collapsible and removes the panic-y API entirely.♻️ One way to keep this collapsible
- 7 if own_user.is_some() - && result.own_pushname.is_none() - && wire_type_raw == wire_type::LENGTH_DELIMITED => - { + 7 if wire_type_raw == wire_type::LENGTH_DELIMITED => { let (len, vlen) = read_varint(&buf[pos..])?; pos += vlen; let end = checked_end(pos, len, buf.len(), "pushname")?; - if let Some(name) = extract_own_pushname(&buf[pos..end], own_user.unwrap()) { + if result.own_pushname.is_none() + && let Some(own_user) = own_user + && let Some(name) = extract_own_pushname(&buf[pos..end], own_user) + { result.own_pushname = Some(name); } pos = end; }As per coding guidelines, "Use let-chains (
if let Some(x) = foo && let Some(y) = x.bar { ... }) instead of nestedif letblocks to maintain collapsible if patterns" and "never use.unwrap()outside tests".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/history_sync.rs` around lines 99 - 107, The branch uses own_user.unwrap() — replace that by moving own_user into the if-let chain so there is no unwrap; e.g. change the condition to bind the user (if let Some(own) = own_user && result.own_pushname.is_none() && wire_type_raw == wire_type::LENGTH_DELIMITED) then call extract_own_pushname(&buf[pos..end], own) (keep the existing read_varint and checked_end usage), ensuring the arm remains collapsible and panics are avoided.
♻️ Duplicate comments (2)
wacore/src/types/events.rs (2)
79-94:⚠️ Potential issue | 🟠 MajorDon't do the first full
HistorySyncdecode on the caller thread.
get()still runswa::HistorySync::decode(...)inline insideOnceLock::get_or_init, so the first async handler that touches a large blob pays the full parse cost on a Tokio worker. That keeps the lazy API, but it reintroduces scheduler stalls on exactly the hot path this change is trying to avoid.As per coding guidelines, "All I/O uses Tokio; wrap blocking I/O (
ureq) and heavy CPU work intokio::task::spawn_blocking".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 79 - 94, get() performs the heavy wa::HistorySync::decode(...) synchronously inside OnceLock::get_or_init, blocking the caller thread; change get() to be async (pub async fn get(&self) -> Option<&wa::HistorySync>) and do a fast-path check of self.parsed.get() returning if present, otherwise offload the decode to tokio::task::spawn_blocking (spawn_blocking(move || wa::HistorySync::decode(&raw_bytes[..]).ok().map(Box::new))) await the join, then store the boxed result into the OnceLock using parsed.set(...) (ignore the Err case if another task beat us) and finally return parsed.get().as_deref(); reference the symbols get(), parsed, OnceLock::set/get(), and wa::HistorySync::decode when making the change.
169-180:⚠️ Potential issue | 🟠 MajorKeep
Serializedeterministic and metadata-only.This output still depends on handler order: if another consumer called
.get()first,datasuddenly contains the full decoded blob; otherwise it serializes asnull. On a sharedArc<Event>, that makes logs/telemetry nondeterministic and can unexpectedly dump full history contents into some sinks but not others.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 169 - 180, The Serialize impl for LazyHistorySync currently calls self.parsed.get() which can trigger decoding and makes serialization nondeterministic; change the serialization to be metadata-only and deterministic by not touching parsed.get(): in the Serialize impl for LazyHistorySync (the serialize method) remove the call to self.parsed.get().and_then(...) and instead serialize a stable metadata value (e.g. serialize_field("data", &Option::<()>::None) or a boolean flag that is obtained via a non-decoding check API if one exists) so the serializer never invokes decoding and always produces deterministic output.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@wacore/src/history_sync.rs`:
- Around line 55-63: The current code only sets initial Vec capacity (estimated)
so ZlibDecoder.read_to_end can still expand beyond 64MB and OOM; change to
enforce a hard decompressed-size limit by (1) capping the reserved capacity with
the hard limit (use estimated.min(MAX_DECOMPRESSED) when building decompressed)
and (2) wrapping the ZlibDecoder with Read::take(MAX_DECOMPRESSED as u64) (e.g.,
let mut limited = decoder.take(MAX); limited.read_to_end(&mut decompressed)?)
and after reading, if the decoder would have more data (or decompressed.len() ==
MAX) return an error; use a named constant like MAX_DECOMPRESSED = 64 * 1024 *
1024 and reference variables/functions estimated, decompressed, ZlibDecoder, and
read_to_end when locating the change.
---
Outside diff comments:
In `@src/history_sync.rs`:
- Around line 256-277: The current read/modify/write in history_sync.rs using
backend.get_tc_token and backend.put_tc_token allows a race that can overwrite a
newer sender_timestamp; change this to perform the max-merge inside a single
backend operation or transaction. Add or use an atomic backend method (e.g.,
backend.merge_tc_token or a transactional put method) that takes token_key and
the incoming candidate/tc values and computes the merged TcTokenEntry (max of
existing.sender_timestamp and candidate.tc_token_sender_timestamp, and
appropriate token_timestamp logic) inside the backend implementation, then
persist it in one step; update the call site in history_sync.rs to call that
atomic method instead of calling get_tc_token followed by put_tc_token so the
sender_timestamp merge is guaranteed atomic.
In `@wacore/src/history_sync.rs`:
- Around line 99-107: The branch uses own_user.unwrap() — replace that by moving
own_user into the if-let chain so there is no unwrap; e.g. change the condition
to bind the user (if let Some(own) = own_user && result.own_pushname.is_none()
&& wire_type_raw == wire_type::LENGTH_DELIMITED) then call
extract_own_pushname(&buf[pos..end], own) (keep the existing read_varint and
checked_end usage), ensuring the arm remains collapsible and panics are avoided.
---
Duplicate comments:
In `@wacore/src/types/events.rs`:
- Around line 79-94: get() performs the heavy wa::HistorySync::decode(...)
synchronously inside OnceLock::get_or_init, blocking the caller thread; change
get() to be async (pub async fn get(&self) -> Option<&wa::HistorySync>) and do a
fast-path check of self.parsed.get() returning if present, otherwise offload the
decode to tokio::task::spawn_blocking (spawn_blocking(move ||
wa::HistorySync::decode(&raw_bytes[..]).ok().map(Box::new))) await the join,
then store the boxed result into the OnceLock using parsed.set(...) (ignore the
Err case if another task beat us) and finally return parsed.get().as_deref();
reference the symbols get(), parsed, OnceLock::set/get(), and
wa::HistorySync::decode when making the change.
- Around line 169-180: The Serialize impl for LazyHistorySync currently calls
self.parsed.get() which can trigger decoding and makes serialization
nondeterministic; change the serialization to be metadata-only and deterministic
by not touching parsed.get(): in the Serialize impl for LazyHistorySync (the
serialize method) remove the call to self.parsed.get().and_then(...) and instead
serialize a stable metadata value (e.g. serialize_field("data",
&Option::<()>::None) or a boolean flag that is obtained via a non-decoding check
API if one exists) so the serializer never invokes decoding and always produces
deterministic output.
🪄 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: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: ab221560-ee1d-4f54-af46-c197e205c199
📒 Files selected for processing (3)
src/history_sync.rswacore/src/history_sync.rswacore/src/types/events.rs
- Add hard 64MB decompression limit via Read::take() to prevent OOM - Fix u64->usize truncation in extract_own_pushname (use try_from) - Remove own_user.unwrap() — move into if-let chain - Remove dead code: into_raw_bytes, conversations_raw, ConversationIterator (zero callers, no tests — re-add when needed) - Revert pub(crate) on wire helpers back to private - Take ownership of TcTokenCandidate in storage loop to avoid tc_token.clone() - Make Serialize metadata-only (sync_type, chunk_order, progress) to avoid non-deterministic output depending on handler execution order
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/history_sync.rs (1)
149-176:⚠️ Potential issue | 🟠 MajorUse decompressed size estimate, not compressed size, for deciding whether to spawn_blocking.
The
<256 KiBcompressed threshold does not bound the actual decompression work. A highly compressible blob staying under 256 KiB compressed can still decompress to tens of MB and consume significant CPU for zlib decompression and protobuf field scanning on the async runtime. Per coding guidelines, wrap heavy CPU work intokio::task::spawn_blocking. Instead, key the decision on the decompressed size estimate (already computed via the 4× multiplier) or unconditionally usespawn_blockingfor all blobs.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@src/history_sync.rs` around lines 149 - 176, The decision to call spawn_blocking currently uses the compressed size (INLINE_THRESHOLD) and can let expensive decompression/protobuf work run on the async runtime; change the branching in the block around INLINE_THRESHOLD to instead estimate decompressed work (use the existing compressed_size_hint or compute a decompressed_estimate from compressed_data.len() and the 4× multiplier) and call self.runtime.spawn_blocking for cases where decompressed_estimate >= INLINE_THRESHOLD (or simply always use spawn_blocking), moving the process_history_sync invocation into the blocking closure and sending the result over the oneshot as before; update references to INLINE_THRESHOLD, compressed_data, compressed_size_hint, process_history_sync, and self.runtime.spawn_blocking accordingly.
♻️ Duplicate comments (3)
wacore/src/history_sync.rs (1)
65-67:⚠️ Potential issue | 🟠 MajorFail closed when the decompressed stream exceeds the cap.
take(MAX_DECOMPRESSED)stops after 64 MiB, but it does not tell you whether the zlib stream actually ended there. A blob that expands past the cap is currently truncated and then parsed as if it were complete. Read one extra byte and reject anything larger than the limit.🛡️ Minimal fix
- let decoder = ZlibDecoder::new(compressed_data.as_slice()); - let mut limited = decoder.take(MAX_DECOMPRESSED); + let decoder = ZlibDecoder::new(compressed_data.as_slice()); + let mut limited = decoder.take(MAX_DECOMPRESSED + 1); limited.read_to_end(&mut decompressed)?; + if decompressed.len() as u64 > MAX_DECOMPRESSED { + return Err(HistorySyncError::MalformedProtobuf( + "history sync exceeds decompressed size limit".into(), + )); + }🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/history_sync.rs` around lines 65 - 67, The decompression currently uses decoder.take(MAX_DECOMPRESSED) and blindly treats the truncated read as a full stream; fix this by reading into decompressed via the existing limited reader, then attempt to read one more byte from the underlying decoder to detect overflow and return an error if any extra data is present. Concretely, after limited.read_to_end(&mut decompressed) on the ZlibDecoder (decoder) and using the MAX_DECOMPRESSED cap, try reading one extra byte (e.g., into a single-byte buffer) from decoder (or from limited if appropriate) and if that read returns Ok(n) with n > 0 or would block/indicate more data, return an error instead of proceeding; keep using the same symbols decoder, limited, MAX_DECOMPRESSED, and decompressed to locate where to insert this check.wacore/src/types/events.rs (2)
96-99: 🛠️ Refactor suggestion | 🟠 MajorReturn
Bytesfrom the public raw-bytes accessor.Exposing
&[u8]forces a copy for any consumer that needs to retain or forward the blob after the event borrow ends, which undermines the zero-copy handoff this type is meant to provide.♻️ Suggested API shape
- pub fn raw_bytes(&self) -> &[u8] { - &self.raw_bytes + pub fn raw_bytes(&self) -> Bytes { + self.raw_bytes.clone() }If you still want a borrowed view, add a separate
as_raw_bytes(&self) -> &[u8].🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 96 - 99, Change the public accessor raw_bytes(&self) -> &[u8] to return an owned Bytes (from the bytes crate) so callers can retain/forward the blob without copying; keep a borrowed view by adding as_raw_bytes(&self) -> &[u8] if needed. Specifically, update the method signature raw_bytes to return bytes::Bytes (constructing it from the internal buffer without additional copies if possible) and add an as_raw_bytes(&self) -> &[u8] that returns &self.raw_bytes for consumers that only need a borrowed slice.
79-94:⚠️ Potential issue | 🟠 MajorOffload the first full protobuf decode.
get()runswa::HistorySync::decode()inline insideOnceLock::get_or_init(). Any async handler that touches a largeEvent::HistorySynccan burn that entire decode on a Tokio worker thread. Please add an offloaded initializer/accessor and keepget()as a cheap read once initialized. As per coding guidelines, "All I/O uses Tokio; wrap blocking I/O (ureq) and heavy CPU work intokio::task::spawn_blocking".🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@wacore/src/types/events.rs` around lines 79 - 94, The current get() runs wa::HistorySync::decode() inside OnceLock::get_or_init which performs heavy CPU work on the caller thread; add an async initializer that offloads decoding via tokio::task::spawn_blocking and keep get() as a cheap reader. Concretely: leave pub fn get(&self) -> Option<&wa::HistorySync> as a simple parsed.get().as_deref(), and add an async method (e.g. pub async fn ensure_parsed_blocking(&self) -> Option<&wa::HistorySync>) that checks if self.parsed.get().is_none(), calls tokio::task::spawn_blocking(move || wa::HistorySync::decode(&self.raw_bytes[..]).ok().map(Box::new)).await, and if decoding succeeded uses self.parsed.set(decoded_box) to populate the OnceLock; finally return self.parsed.get().as_deref(). Use the existing fields parsed, raw_bytes and the OnceLock API and ensure spawn_blocking handles the heavy decode so callers can await ensure_parsed_blocking before calling get().
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@src/history_sync.rs`:
- Around line 149-176: The decision to call spawn_blocking currently uses the
compressed size (INLINE_THRESHOLD) and can let expensive decompression/protobuf
work run on the async runtime; change the branching in the block around
INLINE_THRESHOLD to instead estimate decompressed work (use the existing
compressed_size_hint or compute a decompressed_estimate from
compressed_data.len() and the 4× multiplier) and call
self.runtime.spawn_blocking for cases where decompressed_estimate >=
INLINE_THRESHOLD (or simply always use spawn_blocking), moving the
process_history_sync invocation into the blocking closure and sending the result
over the oneshot as before; update references to INLINE_THRESHOLD,
compressed_data, compressed_size_hint, process_history_sync, and
self.runtime.spawn_blocking accordingly.
---
Duplicate comments:
In `@wacore/src/history_sync.rs`:
- Around line 65-67: The decompression currently uses
decoder.take(MAX_DECOMPRESSED) and blindly treats the truncated read as a full
stream; fix this by reading into decompressed via the existing limited reader,
then attempt to read one more byte from the underlying decoder to detect
overflow and return an error if any extra data is present. Concretely, after
limited.read_to_end(&mut decompressed) on the ZlibDecoder (decoder) and using
the MAX_DECOMPRESSED cap, try reading one extra byte (e.g., into a single-byte
buffer) from decoder (or from limited if appropriate) and if that read returns
Ok(n) with n > 0 or would block/indicate more data, return an error instead of
proceeding; keep using the same symbols decoder, limited, MAX_DECOMPRESSED, and
decompressed to locate where to insert this check.
In `@wacore/src/types/events.rs`:
- Around line 96-99: Change the public accessor raw_bytes(&self) -> &[u8] to
return an owned Bytes (from the bytes crate) so callers can retain/forward the
blob without copying; keep a borrowed view by adding as_raw_bytes(&self) ->
&[u8] if needed. Specifically, update the method signature raw_bytes to return
bytes::Bytes (constructing it from the internal buffer without additional copies
if possible) and add an as_raw_bytes(&self) -> &[u8] that returns
&self.raw_bytes for consumers that only need a borrowed slice.
- Around line 79-94: The current get() runs wa::HistorySync::decode() inside
OnceLock::get_or_init which performs heavy CPU work on the caller thread; add an
async initializer that offloads decoding via tokio::task::spawn_blocking and
keep get() as a cheap reader. Concretely: leave pub fn get(&self) ->
Option<&wa::HistorySync> as a simple parsed.get().as_deref(), and add an async
method (e.g. pub async fn ensure_parsed_blocking(&self) ->
Option<&wa::HistorySync>) that checks if self.parsed.get().is_none(), calls
tokio::task::spawn_blocking(move ||
wa::HistorySync::decode(&self.raw_bytes[..]).ok().map(Box::new)).await, and if
decoding succeeded uses self.parsed.set(decoded_box) to populate the OnceLock;
finally return self.parsed.get().as_deref(). Use the existing fields parsed,
raw_bytes and the OnceLock API and ensure spawn_blocking handles the heavy
decode so callers can await ensure_parsed_blocking before calling get().
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: ASSERTIVE
Plan: Pro
Run ID: 74a4653c-ef9c-4448-904f-2c762d49b4ca
📒 Files selected for processing (3)
src/history_sync.rswacore/src/history_sync.rswacore/src/types/events.rs
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: bd71f0fe17
ℹ️ 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".
| let mut limited = decoder.take(MAX_DECOMPRESSED); | ||
| limited.read_to_end(&mut decompressed)?; |
There was a problem hiding this comment.
Using Read::take(MAX_DECOMPRESSED) here silently truncates history-sync payloads above 64 MiB, because take() reports EOF when the limit is hit. That means oversized but otherwise valid blobs can be partially parsed and emitted as if complete (or fail only after the protocol receipt was already sent), causing irreversible data loss for large accounts. Please explicitly detect limit exhaustion and return a hard error for over-cap blobs instead of continuing with truncated bytes.
Useful? React with 👍 / 👎.
| if let Some(parts) = wacore_binary::jid::parse_jid_fast(&conv.id) | ||
| && (parts.server == "g.us" || parts.server == "newsletter" || parts.server == "bot") | ||
| { |
There was a problem hiding this comment.
This new fast-path filter only excludes server == "bot", but previous logic excluded all jid.is_bot() values, including phone-number bots on s.whatsapp.net (e.g. 1313555* / 131655500*). Because store_tc_token_candidate no longer re-checks is_bot, tc tokens from those bot chats can now be cached and later reused on sends, which changes privacy-token behavior for bot conversations.
Useful? React with 👍 / 👎.
1 participant
Summary
Replaces
Event::JoinedGroup(LazyConversation)(per-conversation, misleading name) and deadEvent::HistorySync(HistorySync)with a singleEvent::HistorySync(Box<LazyHistorySync>)per sync blob, plus performance optimizations.Consumer API
Changes
Event refactor
Event::JoinedGroup(LazyConversation)removed — name was wrong (carried DMs, groups, newsletters)Event::HistorySync(HistorySync)removed — dead code, never dispatchedEvent::HistorySync(Box<LazyHistorySync>)— single event per sync blob with lazy decodePerformance
async_channel/oneshot/secondary spawnBytesonly kept when event listeners existspawn_blockingparse_jid_fastearly-exit — groups/newsletters/bots filtered without JID allocationRead::take()to prevent OOMBox<wa::HistorySync>in OnceLock — avoids ~500 bytes inline reservationBytescheaply, creates freshOnceLock(no accidental deep copy)TcTokenEntrySafety
Read::take()prevents OOM on malformed blobsu64 -> usizeconversion viatry_from(no truncation on 32-bit/WASM)own_user.unwrap()eliminated — moved into if-let chainWhat's preserved
has_handlers()gates blob retention and event dispatch)Breaking changes
Event::JoinedGroupremovedEvent::HistorySyncpayload changed fromwa::HistorySynctoBox<LazyHistorySync>LazyConversationremovedTest plan
cargo clippy --all --tests— zero warningscargo fmt --all— cleanJoinedGroup,LazyConversation, or old channel code