From 532527b8c793b4e6e88b3161860679b9fa35a45d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 15:06:32 -0300 Subject: [PATCH 1/9] perf(history-sync)!: store the compressed payload and expose a streaming 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 --- src/history_sync.rs | 71 +- wacore/benches/history_sync_benchmark.rs | 27 +- wacore/binary/src/zlib_pool.rs | 14 +- wacore/src/history_sync.rs | 1007 +++++++++++++++++----- wacore/src/types/events.rs | 340 ++++---- 5 files changed, 1066 insertions(+), 393 deletions(-) diff --git a/src/history_sync.rs b/src/history_sync.rs index f9f2b657d..e57611b67 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -93,12 +93,6 @@ impl Client { return; } - // file_length is the decrypted (but still zlib-compressed) blob size, not - // the final decompressed size. We still pass it as a hint — the decompressor - // uses it with a 4x multiplier, which is a better estimate than guessing - // from the encrypted size (which includes MAC/padding overhead). - let compressed_size_hint = notification.file_length.filter(|&s| s > 0); - // Use take() to avoid cloning large payloads - moves ownership instead let compressed_data = if let Some(inline_payload) = notification.initial_hist_bootstrap_inline_payload.take() @@ -161,17 +155,12 @@ impl Client { compressed_data, own_user.as_deref(), retain_history_blob, - compressed_size_hint, )) } else { let (result_tx, result_rx) = futures::channel::oneshot::channel(); let blocking_fut = self.runtime.spawn_blocking(Box::new(move || { - let result = process_history_sync( - compressed_data, - own_user.as_deref(), - retain_history_blob, - compressed_size_hint, - ); + let result = + process_history_sync(compressed_data, own_user.as_deref(), retain_history_blob); let _ = result_tx.send(result); })); self.runtime @@ -225,9 +214,10 @@ impl Client { self.store_history_sync_msg_secrets(sync_result.msg_secret_records) .await; - if let Some(decompressed) = sync_result.decompressed_bytes { + if let Some(compressed) = sync_result.compressed_bytes { let lazy_hs = LazyHistorySync::new( - decompressed, + compressed, + sync_result.decompressed_size, notification.sync_type().into(), notification.chunk_order, notification.progress, @@ -570,6 +560,57 @@ mod tests { assert_eq!(got, Some(secret)); } + #[tokio::test] + async fn process_history_sync_task_dispatches_compressed_lazy_event() { + let client = crate::test_utils::create_test_client_with_name("history_lazy_event").await; + client.is_running.store(true, Ordering::Relaxed); + + let chat = "5511777776666@s.whatsapp.net"; + let history_sync = wa::HistorySync { + sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32, + conversations: vec![wa::Conversation { + id: chat.to_string(), + ..Default::default() + }], + ..Default::default() + }; + let raw_len = history_sync.encode_to_vec().len(); + let compressed = compress_history_sync(&history_sync); + let compressed_copy = compressed.clone(); + let notification = HistorySyncNotification { + file_length: Some(compressed.len() as u64), + sync_type: Some(wa::message::HistorySyncType::InitialBootstrap as i32), + initial_hist_bootstrap_inline_payload: Some(compressed), + ..Default::default() + }; + + // Register a handler BEFORE the task so retain_blob is true. + let (handler, event_rx) = wacore::types::events::ChannelEventHandler::new(); + client.core.event_bus.add_handler(handler); + + client + .process_history_sync_task("HIST_LAZY_EVENT".to_string(), notification) + .await; + + let event = event_rx.try_recv().expect("HistorySync event dispatched"); + let crate::types::events::Event::HistorySync(lazy) = &*event else { + panic!("expected HistorySync event, got {event:?}"); + }; + + // The event carries the original compressed payload plus the exact + // inflated size, and every consumption path works. + assert_eq!(lazy.compressed_bytes().as_ref(), &compressed_copy[..]); + assert_eq!(lazy.decompressed_size(), raw_len); + let decoded = lazy.get().expect("decodes"); + assert_eq!(decoded.conversations[0].id, chat); + let mut stream = lazy.stream(); + assert_eq!( + stream.next_conversation().unwrap().unwrap().id, + chat, + "stream still works after get()" + ); + } + #[tokio::test] async fn process_history_sync_task_stores_bot_dm_secret_alias() { let client = diff --git a/wacore/benches/history_sync_benchmark.rs b/wacore/benches/history_sync_benchmark.rs index 0cc73cb2b..2e4e1ea6c 100644 --- a/wacore/benches/history_sync_benchmark.rs +++ b/wacore/benches/history_sync_benchmark.rs @@ -103,14 +103,33 @@ fn bench_process_history_sync(bencher: divan::Bencher) { bencher .with_inputs(setup_history_sync_blob) .bench_values(|blob| { - // retain_blob = true exercises the full-buffer path. The result - // (records + retained blob) is returned so the harness drops it - // outside the measured window, like a consumer would later. + // retain_blob = true also hands the compressed input back. The + // result (records + retained blob) is returned so the harness + // drops it outside the measured window, like a consumer would. black_box(wacore::history_sync::process_history_sync( black_box(blob), None, true, - None, )) }); } + +/// Consumer-side pass over the retained blob: drain every conversation through +/// the public stream and decode the remainder, the path an Event::HistorySync +/// handler pays per chunk. +#[divan::bench(sample_count = 5)] +fn bench_history_sync_stream_drain(bencher: divan::Bencher) { + bencher + .with_inputs(setup_history_sync_blob) + .bench_values(|blob| { + let mut stream = wacore::history_sync::HistorySyncStream::new( + black_box(&blob), + wacore::history_sync::MAX_DECOMPRESSED, + ); + let mut messages = 0usize; + while let Some(conversation) = stream.next_conversation().unwrap() { + messages += conversation.messages.len(); + } + black_box((messages, stream.remainder().unwrap())) + }); +} diff --git a/wacore/binary/src/zlib_pool.rs b/wacore/binary/src/zlib_pool.rs index c55d43b00..7cd30d5df 100644 --- a/wacore/binary/src/zlib_pool.rs +++ b/wacore/binary/src/zlib_pool.rs @@ -92,6 +92,12 @@ impl<'a> InflateReader<'a> { self.eof && self.cursor >= self.buf.len() } + /// Total decompressed bytes produced so far. After the stream ends this is + /// the blob's exact inflated size. + pub fn total_out(&self) -> u64 { + self.total_out + } + fn pump(&mut self) -> io::Result<()> { // Drop the consumed prefix before growing, so the buffer holds roughly // just the record currently being accumulated. @@ -225,8 +231,12 @@ pub fn decompress_zlib_pooled(compressed: &[u8], max_size: u64) -> io::Result, @@ -27,9 +39,13 @@ pub struct HistorySyncResult { /// Tctoken candidates extracted from 1:1 conversations during streaming. pub tc_token_candidates: Vec, pub msg_secret_records: Vec, - /// The full decompressed protobuf blob, only retained when event - /// listeners exist. Wrapped in `LazyHistorySync` for on-demand decoding. - pub decompressed_bytes: Option, + /// The original zlib-compressed input, handed back (moved, never copied or + /// re-inflated) only when event listeners exist. Wrapped in + /// `LazyHistorySync` for on-demand consumption. + pub compressed_bytes: Option, + /// Exact size of the fully inflated blob, counted during the extraction + /// walk. Carried into `LazyHistorySync` as the inflate cap. + pub decompressed_size: usize, } mod wire_type { @@ -43,248 +59,332 @@ mod wire_type { /// Decompress and process a history sync blob. /// -/// **Memory strategy**: Decompresses the entire blob into a single `Bytes` -/// buffer, then scans top-level fields and partially decodes only the -/// conversation fields needed for internal caches. -/// -/// After decompression, the compressed input is dropped immediately, so peak -/// memory = max(compressed, decompressed) + small overhead, not both. +/// **Memory strategy**: always streams — inflates with a bounded window and +/// extracts each top-level field as soon as its bytes are buffered, so peak +/// memory is the largest single conversation, never the whole blob. With +/// `retain_blob`, the original compressed input is handed back in +/// [`HistorySyncResult::compressed_bytes`] (a move, no copy and no second +/// inflate) for on-demand consumer decoding via `LazyHistorySync`. pub fn process_history_sync( compressed_data: Vec, own_user: Option<&str>, retain_blob: bool, - _compressed_size_hint: Option, ) -> Result { - // Hard limit to prevent OOM on malformed blobs. - // Typical InitialBootstrap: 5-20 MB decompressed. - const MAX_DECOMPRESSED: u64 = 64 * 1024 * 1024; - - // When the caller doesn't need the full decompressed blob (no Event::HistorySync - // consumer), stream-decompress and extract incrementally so peak memory stays - // ~one conversation instead of the whole blob. - if !retain_blob { - return process_history_sync_streaming(&compressed_data, own_user, MAX_DECOMPRESSED); + let mut result = process_history_sync_streaming(&compressed_data, own_user, MAX_DECOMPRESSED)?; + if retain_blob { + result.compressed_bytes = Some(Bytes::from(compressed_data)); } + Ok(result) +} - let decompressed = decompress_zlib_pooled(&compressed_data, MAX_DECOMPRESSED) - .map_err(HistorySyncError::DecompressionError)?; - drop(compressed_data); - - let buf = Bytes::from(decompressed); - let mut pos = 0; - let mut result = HistorySyncResult { - own_pushname: None, - nct_salt: None, - conversations_processed: 0, - tc_token_candidates: Vec::new(), - // Grown on demand: a full pre-count pass scanned the whole blob just to - // size a Vec that only holds the secret-record subset (it over-allocated - // and cost ~2.5% of the decode); plain growth is cheaper here. - msg_secret_records: Vec::new(), - // Always retained on this path: the `!retain_blob` case returned above - // and ran the streaming variant, so control only reaches here when the - // caller wants the blob. - decompressed_bytes: Some(buf.clone()), - }; - - while pos < buf.len() { - let (tag, bytes_read) = read_varint(&buf[pos..])?; - pos += bytes_read; - - let field_number = (tag >> 3) as u32; - let wire_type_raw = (tag & 0x7) as u32; - - match field_number { - // conversations (repeated, length-delimited) - tags::history_sync::CONVERSATIONS 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(), "conversation")?; - - result.conversations_processed += 1; - if let Some(candidate) = - extract_conversation_fields(&buf[pos..end], &mut result.msg_secret_records) - { - result.tc_token_candidates.push(candidate); - } - pos = end; - } - - // pushnames (repeated, length-delimited). - // Uses `Option::is_some()` in the guard rather than an - // `if let` guard — the latter requires Rust 1.94+. The inner - // `if let` is the defensive complement: if the guard's - // invariant is ever weakened by a future refactor, we skip - // the arm body instead of panicking. - tags::history_sync::PUSHNAMES - if own_user.is_some() - && result.own_pushname.is_none() - && 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(own) = own_user - && let Some(name) = extract_own_pushname(&buf[pos..end], own) - { - result.own_pushname = Some(name); - } - pos = end; - } - - // nctSalt (optional bytes, length-delimited) - // Delivered during initial pairing so cstoken is available immediately. - // Source: storeNctSaltFromHistorySync in WAWeb/History/MsgHandlerAction.js - tags::history_sync::NCT_SALT 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(), "nctSalt")?; +/// One top-level protobuf field, borrowed from the walker's inflate window. +struct RawField<'w> { + field_number: u32, + wire_type: u32, + /// Full wire span (tag + length prefix + payload): what a raw re-emit of + /// the field must copy. + raw: &'w [u8], + /// Offset of the payload inside `raw`. Only meaningful for + /// length-delimited fields. + payload_start: usize, +} - let salt = buf[pos..end].to_vec(); - if !salt.is_empty() { - result.nct_salt = Some(salt); - } - pos = end; - } +/// The single protobuf wire walk over a compressed HistorySync blob: an +/// incremental inflate window plus top-level field framing. Both the internal +/// extractor ([`process_history_sync`]) and the public [`HistorySyncStream`] +/// consume it, so the format knowledge lives in exactly one place. +struct FieldWalker<'a> { + reader: InflateReader<'a>, + /// Span of the previously yielded field, consumed lazily on the next call + /// so the caller can keep borrowing the field bytes from the window in + /// between. + pending: usize, +} - _ => { - pos = skip_field(wire_type_raw, &buf, pos)?; - } +impl<'a> FieldWalker<'a> { + fn new(compressed: &'a [u8], max_decompressed: u64) -> Self { + Self { + reader: InflateReader::new(compressed, max_decompressed), + pending: 0, } } - Ok(result) -} + fn total_out(&self) -> u64 { + self.reader.total_out() + } -/// Streaming variant of [`process_history_sync`] for when the full decompressed -/// blob is NOT needed (`retain_blob == false`). Decompresses incrementally and -/// parses each top-level field as soon as its bytes are buffered, so peak memory -/// is bounded by the largest single conversation rather than the whole blob. -/// Produces the same extraction results (secrets, tctokens, pushname, nctSalt) -/// as the full path, but with `decompressed_bytes == None`. -fn process_history_sync_streaming( - compressed_data: &[u8], - own_user: Option<&str>, - max_decompressed: u64, -) -> Result { - let mut reader = InflateReader::new(compressed_data, max_decompressed); - let mut result = HistorySyncResult { - own_pushname: None, - nct_salt: None, - conversations_processed: 0, - tc_token_candidates: Vec::new(), - msg_secret_records: Vec::new(), - decompressed_bytes: None, - }; + /// Re-borrow the payload of the field most recently yielded by + /// [`FieldWalker::next_field`] (it stays buffered until the next call). + fn pending_payload(&self, payload_start: usize) -> &[u8] { + &self.reader.available()[payload_start..self.pending] + } + + /// Advance to the next top-level field, returning `Ok(None)` at clean EOF. + /// The returned borrows point into the inflate window and stay valid until + /// the next `next_field` call. + fn next_field(&mut self) -> Result>, HistorySyncError> { + self.reader.consume(self.pending); + self.pending = 0; - loop { // A field starts with a tag varint; stop cleanly when the stream ends. - if !reader + if !self + .reader .ensure(1) .map_err(HistorySyncError::DecompressionError)? { - break; + return Ok(None); } // A varint is at most 10 bytes (fewer is fine right at EOF). - reader + self.reader .ensure(10) .map_err(HistorySyncError::DecompressionError)?; - let (tag, tlen) = read_varint(reader.available())?; - reader.consume(tlen); - + let (tag, tlen) = read_varint(self.reader.available())?; let field_number = (tag >> 3) as u32; let wire_type_raw = (tag & 0x7) as u32; - match wire_type_raw { + let (span, payload_start) = match wire_type_raw { wire_type::LENGTH_DELIMITED => { - reader - .ensure(10) + self.reader + .ensure(tlen + 10) .map_err(HistorySyncError::DecompressionError)?; - let (len, vlen) = read_varint(reader.available())?; - reader.consume(vlen); + let (len, vlen) = read_varint(&self.reader.available()[tlen..])?; let len = usize::try_from(len).map_err(|_| { HistorySyncError::MalformedProtobuf(format!( "field length overflows usize: {len}" )) })?; - if !reader - .ensure(len) + let payload_start = tlen + vlen; + let span = payload_start.checked_add(len).ok_or_else(|| { + HistorySyncError::MalformedProtobuf(format!( + "field span overflows: header={payload_start}, len={len}" + )) + })?; + if !self + .reader + .ensure(span) .map_err(HistorySyncError::DecompressionError)? { return Err(HistorySyncError::MalformedProtobuf( "length-delimited field truncated".into(), )); } - { - let value = &reader.available()[..len]; - match field_number { - // conversations (repeated) - tags::history_sync::CONVERSATIONS => { - result.conversations_processed += 1; - if let Some(candidate) = - extract_conversation_fields(value, &mut result.msg_secret_records) - { - result.tc_token_candidates.push(candidate); - } - } - // pushnames (repeated) — only our own is needed - tags::history_sync::PUSHNAMES => { - if result.own_pushname.is_none() - && let Some(own) = own_user - && let Some(name) = extract_own_pushname(value, own) - { - result.own_pushname = Some(name); - } - } - tags::history_sync::NCT_SALT if !value.is_empty() => { - result.nct_salt = Some(value.to_vec()); - } - _ => {} - } - } - reader.consume(len); + (span, payload_start) } wire_type::VARINT => { - reader - .ensure(10) + self.reader + .ensure(tlen + 10) .map_err(HistorySyncError::DecompressionError)?; - let (_, vlen) = read_varint(reader.available())?; - reader.consume(vlen); + let (_, vlen) = read_varint(&self.reader.available()[tlen..])?; + (tlen + vlen, tlen) } wire_type::FIXED64 => { - if !reader - .ensure(8) + if !self + .reader + .ensure(tlen + 8) .map_err(HistorySyncError::DecompressionError)? { return Err(HistorySyncError::MalformedProtobuf( "fixed64 field truncated".into(), )); } - reader.consume(8); + (tlen + 8, tlen) } wire_type::FIXED32 => { - if !reader - .ensure(4) + if !self + .reader + .ensure(tlen + 4) .map_err(HistorySyncError::DecompressionError)? { return Err(HistorySyncError::MalformedProtobuf( "fixed32 field truncated".into(), )); } - reader.consume(4); + (tlen + 4, tlen) } _ => { return Err(HistorySyncError::MalformedProtobuf(format!( "unknown wire type {wire_type_raw}" ))); } + }; + + self.pending = span; + Ok(Some(RawField { + field_number, + wire_type: wire_type_raw, + raw: &self.reader.available()[..span], + payload_start, + })) + } +} + +/// The internal extraction pass: decompresses incrementally and pulls out +/// secrets, tctokens, pushname and nctSalt as each top-level field is +/// buffered, so peak memory is bounded by the largest single conversation +/// rather than the whole blob. +fn process_history_sync_streaming( + compressed_data: &[u8], + own_user: Option<&str>, + max_decompressed: u64, +) -> Result { + let mut walker = FieldWalker::new(compressed_data, max_decompressed); + let mut result = HistorySyncResult { + own_pushname: None, + nct_salt: None, + conversations_processed: 0, + tc_token_candidates: Vec::new(), + // Grown on demand: a full pre-count pass scanned the whole blob just to + // size a Vec that only holds the secret-record subset (it over-allocated + // and cost ~2.5% of the decode); plain growth is cheaper here. + msg_secret_records: Vec::new(), + compressed_bytes: None, + decompressed_size: 0, + }; + + while let Some(field) = walker.next_field()? { + if field.wire_type != wire_type::LENGTH_DELIMITED { + continue; + } + let value = &field.raw[field.payload_start..]; + match field.field_number { + // conversations (repeated) + tags::history_sync::CONVERSATIONS => { + result.conversations_processed += 1; + if let Some(candidate) = + extract_conversation_fields(value, &mut result.msg_secret_records) + { + result.tc_token_candidates.push(candidate); + } + } + // pushnames (repeated) — only our own is needed + tags::history_sync::PUSHNAMES => { + if result.own_pushname.is_none() + && let Some(own) = own_user + && let Some(name) = extract_own_pushname(value, own) + { + result.own_pushname = Some(name); + } + } + tags::history_sync::NCT_SALT if !value.is_empty() => { + result.nct_salt = Some(value.to_vec()); + } + _ => {} } } + result.decompressed_size = walker.total_out() as usize; Ok(result) } +/// Incremental reader over a compressed HistorySync blob: yields conversations +/// one at a time and the non-conversation fields as a final decoded remainder, +/// without ever materializing the whole decompressed blob. +/// +/// Decompression uses a bounded window, so peak memory is roughly the largest +/// single conversation plus the accumulated non-conversation fields (a few KB +/// on conversation-heavy blobs; effectively the whole — small — blob on +/// conversation-less ones such as PushName-only or nctSalt-only chunks). +/// +/// Cost model: one zlib inflate per full pass. A multi-MB InitialBootstrap +/// chunk takes tens of milliseconds to inflate and decode; inside an async +/// handler, prefer draining the stream in `spawn_blocking` (clone the +/// compressed `Bytes` into the closure) when +/// [`LazyHistorySync::decompressed_size`](crate::types::events::LazyHistorySync::decompressed_size) +/// is large. +pub struct HistorySyncStream<'a> { + walker: FieldWalker<'a>, + /// Raw (tag + payload) bytes of every non-conversation top-level field + /// encountered while iterating, decoded at the end by + /// [`HistorySyncStream::remainder`]. Accumulating raw fields makes wire + /// order irrelevant: conversations interleaved with other fields decode + /// identically to a field-ordered blob. + remainder: Vec, + skipped_conversations: usize, +} + +impl<'a> HistorySyncStream<'a> { + /// Reader over `compressed`, refusing to inflate past `max_decompressed` + /// (use [`MAX_DECOMPRESSED`] when the exact inflated size is unknown). + pub fn new(compressed: &'a [u8], max_decompressed: u64) -> Self { + Self { + walker: FieldWalker::new(compressed, max_decompressed), + remainder: Vec::new(), + skipped_conversations: 0, + } + } + + /// Raw protobuf bytes of the next `conversations` entry, or `Ok(None)` at + /// clean EOF. The borrow points into the inflate window and stays valid + /// until the next call on the stream. + /// + /// Every other top-level field encountered along the way is buffered for + /// [`HistorySyncStream::remainder`]. A truncated field or zlib error is + /// fatal; per-conversation decode leniency lives in + /// [`HistorySyncStream::next_conversation`]. + pub fn next_conversation_bytes(&mut self) -> Result, HistorySyncError> { + let payload_start = loop { + let Some(field) = self.walker.next_field()? else { + return Ok(None); + }; + if field.field_number == tags::history_sync::CONVERSATIONS + && field.wire_type == wire_type::LENGTH_DELIMITED + { + break field.payload_start; + } + self.remainder.extend_from_slice(field.raw); + }; + // Re-borrow outside the loop: the field stays buffered (consumed lazily + // on the next walker call), so the payload slice is still in the window. + Ok(Some(self.walker.pending_payload(payload_start))) + } + + /// Decoded variant of [`HistorySyncStream::next_conversation_bytes`]. + /// LENIENT: a conversation that fails prost decode is skipped and counted + /// in [`HistorySyncStream::skipped_conversations`], not fatal — one + /// corrupt entry doesn't void the rest of the blob. + pub fn next_conversation(&mut self) -> Result, HistorySyncError> { + loop { + match self.next_conversation_bytes()? { + None => return Ok(None), + Some(bytes) => match ::decode(bytes) { + Ok(conversation) => return Ok(Some(conversation)), + Err(e) => { + log::debug!("Skipping undecodable history-sync conversation: {e}"); + self.skipped_conversations += 1; + } + }, + } + } + } + + /// How many conversations [`HistorySyncStream::next_conversation`] skipped + /// because they failed to decode. + pub fn skipped_conversations(&self) -> usize { + self.skipped_conversations + } + + /// Decode the accumulated non-conversation fields (pushnames, mappings, + /// settings, nctSalt, ...) as a conversations-less [`wa::HistorySync`]. + /// + /// Call after `next_conversation*` returned `None`. Calling earlier drains + /// the rest of the stream and fails with + /// [`HistorySyncError::UnreadConversations`] if an unread conversation is + /// found, instead of silently dropping it. + pub fn remainder(mut self) -> Result { + while let Some(field) = self.walker.next_field()? { + if field.field_number == tags::history_sync::CONVERSATIONS + && field.wire_type == wire_type::LENGTH_DELIMITED + { + return Err(HistorySyncError::UnreadConversations); + } + self.remainder.extend_from_slice(field.raw); + } + Ok(waproto::codec::history_sync_decode( + self.remainder.as_slice(), + )?) + } +} + /// Compute `pos + len` with overflow and bounds checking. #[inline(always)] fn checked_end( @@ -2177,7 +2277,7 @@ mod tests { let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); encoder.write_all(&hs).unwrap(); let compressed = encoder.finish().unwrap(); - process_history_sync(compressed, None, false, None) + process_history_sync(compressed, None, false) .unwrap() .msg_secret_records } @@ -2219,7 +2319,7 @@ mod tests { let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); encoder.write_all(&hs).unwrap(); let compressed = encoder.finish().unwrap(); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert!(result.msg_secret_records.is_empty()); } @@ -2333,7 +2433,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert_eq!(result.msg_secret_records.len(), 1); assert!(result.msg_secret_records[0].secret.is_empty()); } @@ -2348,7 +2448,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert_eq!(result.nct_salt, Some(salt)); } @@ -2361,7 +2461,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert!(result.nct_salt.is_none()); } @@ -2380,7 +2480,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, Some("0000000000"), false, None).unwrap(); + let result = process_history_sync(compressed, Some("0000000000"), false).unwrap(); assert_eq!(result.nct_salt, Some(salt)); assert_eq!(result.own_pushname.as_deref(), Some("TestUser")); @@ -2436,7 +2536,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert_eq!(result.msg_secret_records.len(), 2); assert_eq!(&*result.msg_secret_records[0].chat_id, chat); @@ -2495,7 +2595,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert_eq!(result.msg_secret_records.len(), 1); assert_eq!(result.msg_secret_records[0].msg_id, "HIST_BOTH"); @@ -2552,7 +2652,7 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert!(result.msg_secret_records.is_empty()); } @@ -2613,24 +2713,86 @@ mod tests { }; let compressed = encode_and_compress(&hs); - let result = process_history_sync(compressed, None, false, None).unwrap(); + let result = process_history_sync(compressed, None, false).unwrap(); assert!(result.msg_secret_records.is_empty()); } - /// The streaming path (retain_blob=false) must produce byte-for-byte the same - /// extraction as the full-decompress path (retain_blob=true), across multiple - /// conversations, a >64 KB conversation that spans decompress chunks, a group - /// (no tctoken), pushname and nctSalt. - #[test] - fn streaming_and_full_paths_produce_identical_results() { + /// Differential oracle: the pre-streaming full-buffer walk, kept test-only + /// so the parity check compares the production stream against an + /// independent implementation instead of a second production path. + fn reference_full_walk(decompressed: &[u8], own_user: Option<&str>) -> HistorySyncResult { + let mut pos = 0; + let mut result = HistorySyncResult { + own_pushname: None, + nct_salt: None, + conversations_processed: 0, + tc_token_candidates: Vec::new(), + msg_secret_records: Vec::new(), + compressed_bytes: None, + decompressed_size: decompressed.len(), + }; + + while pos < decompressed.len() { + let (tag, bytes_read) = read_varint(&decompressed[pos..]).unwrap(); + pos += bytes_read; + let field_number = (tag >> 3) as u32; + let wt = (tag & 0x7) as u32; + match field_number { + tags::history_sync::CONVERSATIONS if wt == wire_type::LENGTH_DELIMITED => { + let (len, vlen) = read_varint(&decompressed[pos..]).unwrap(); + pos += vlen; + let end = checked_end(pos, len, decompressed.len(), "conversation").unwrap(); + result.conversations_processed += 1; + if let Some(candidate) = extract_conversation_fields( + &decompressed[pos..end], + &mut result.msg_secret_records, + ) { + result.tc_token_candidates.push(candidate); + } + pos = end; + } + tags::history_sync::PUSHNAMES + if own_user.is_some() + && result.own_pushname.is_none() + && wt == wire_type::LENGTH_DELIMITED => + { + let (len, vlen) = read_varint(&decompressed[pos..]).unwrap(); + pos += vlen; + let end = checked_end(pos, len, decompressed.len(), "pushname").unwrap(); + if let Some(own) = own_user + && let Some(name) = extract_own_pushname(&decompressed[pos..end], own) + { + result.own_pushname = Some(name); + } + pos = end; + } + tags::history_sync::NCT_SALT if wt == wire_type::LENGTH_DELIMITED => { + let (len, vlen) = read_varint(&decompressed[pos..]).unwrap(); + pos += vlen; + let end = checked_end(pos, len, decompressed.len(), "nctSalt").unwrap(); + let salt = decompressed[pos..end].to_vec(); + if !salt.is_empty() { + result.nct_salt = Some(salt); + } + pos = end; + } + _ => { + pos = skip_field(wt, decompressed, pos).unwrap(); + } + } + } + result + } + + /// Multi-conversation fixture: a >64 KB DM (spans decompress chunks, has a + /// tctoken), a group (tctoken must be ignored), pushname and nctSalt. + fn parity_fixture(own: &str) -> wa::HistorySync { use wa::history_sync::HistorySyncType; - let own = "5511000000000"; let dm = "5511777776666@s.whatsapp.net"; let group = "123456789-987654321@g.us"; let participant = "5511888889999@s.whatsapp.net"; - // Big 1:1 conversation (>64 KB decompressed) carrying a tctoken. let mut big_msgs = Vec::new(); for i in 0..1500u32 { big_msgs.push(wa::HistorySyncMsg { @@ -2656,7 +2818,6 @@ mod tests { ..Default::default() }; - // Group conversation: a secret message, but the tctoken must be ignored. let group_conv = wa::Conversation { id: group.to_string(), messages: vec![wa::HistorySyncMsg { @@ -2677,7 +2838,7 @@ mod tests { ..Default::default() }; - let hs = wa::HistorySync { + wa::HistorySync { sync_type: HistorySyncType::InitialBootstrap as i32, conversations: vec![big_conv, group_conv], pushnames: vec![wa::Pushname { @@ -2686,32 +2847,444 @@ mod tests { }], nct_salt: Some(vec![0x01, 0x02, 0x03, 0x04]), ..Default::default() - }; + } + } + /// The production extraction (always streaming) must match the test-only + /// full-buffer reference walk byte for byte, and `retain_blob` must hand + /// the compressed input back with the exact inflated size. + #[test] + fn streaming_extraction_matches_full_buffer_reference() { + let own = "5511000000000"; + let hs = parity_fixture(own); let compressed = encode_and_compress(&hs); - let full = process_history_sync(compressed.clone(), Some(own), true, None).unwrap(); - let streamed = process_history_sync(compressed, Some(own), false, None).unwrap(); + let decompressed = + wacore_binary::zlib_pool::decompress_zlib_pooled(&compressed, MAX_DECOMPRESSED) + .unwrap(); - assert!(full.decompressed_bytes.is_some(), "full path retains blob"); - assert!( - streamed.decompressed_bytes.is_none(), - "streaming path drops blob" - ); - assert_eq!(full.nct_salt, streamed.nct_salt); - assert_eq!(full.own_pushname, streamed.own_pushname); - assert_eq!(full.own_pushname.as_deref(), Some("Me")); + let reference = reference_full_walk(&decompressed, Some(own)); + let streamed = process_history_sync(compressed.clone(), Some(own), false).unwrap(); + let retained = process_history_sync(compressed.clone(), Some(own), true).unwrap(); + + assert!(streamed.compressed_bytes.is_none(), "no-retain drops input"); assert_eq!( - full.conversations_processed, - streamed.conversations_processed + retained.compressed_bytes.as_deref(), + Some(compressed.as_slice()), + "retain hands the original compressed input back" + ); + assert_eq!(streamed.decompressed_size, decompressed.len()); + assert_eq!(retained.decompressed_size, decompressed.len()); + + for result in [&streamed, &retained] { + assert_eq!(result.nct_salt, reference.nct_salt); + assert_eq!(result.own_pushname, reference.own_pushname); + assert_eq!(result.own_pushname.as_deref(), Some("Me")); + assert_eq!( + result.conversations_processed, + reference.conversations_processed + ); + assert_eq!(result.conversations_processed, 2); + assert_eq!(result.tc_token_candidates, reference.tc_token_candidates); + assert_eq!( + result.tc_token_candidates.len(), + 1, + "only the DM has a tctoken" + ); + assert_eq!(result.msg_secret_records, reference.msg_secret_records); + assert_eq!(result.msg_secret_records.len(), 1500 + 1); + } + } + + /// Collecting `next_conversation()` + `remainder()` and stitching them back + /// together must equal one full prost decode of the decompressed blob. + #[test] + fn stream_parity_with_full_prost_decode() { + let own = "5511000000000"; + let hs = parity_fixture(own); + let compressed = encode_and_compress(&hs); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + let mut conversations = Vec::new(); + while let Some(conversation) = stream.next_conversation().unwrap() { + conversations.push(conversation); + } + assert_eq!(stream.skipped_conversations(), 0); + let mut stitched = stream.remainder().unwrap(); + assert!(stitched.conversations.is_empty()); + stitched.conversations = conversations; + + let decompressed = + wacore_binary::zlib_pool::decompress_zlib_pooled(&compressed, MAX_DECOMPRESSED) + .unwrap(); + let full = waproto::codec::history_sync_decode(&decompressed).unwrap(); + assert_eq!(stitched, full); + } + + /// Conversations interleaved with other top-level fields in any wire order + /// must stream identically: the remainder accumulation makes order + /// irrelevant. + #[test] + fn stream_handles_field_order_shuffled_blobs() { + let conv_a = wa::Conversation { + id: "5511111111111@s.whatsapp.net".into(), + ..Default::default() + }; + let conv_b = wa::Conversation { + id: "5511222222222@s.whatsapp.net".into(), + ..Default::default() + }; + let pushname = wa::Pushname { + id: Some("5511000000000".into()), + pushname: Some("Me".into()), + }; + + // pushname, conv A, unknown varint field, nctSalt, conv B. + let mut blob = Vec::new(); + emit_len_field( + &mut blob, + tags::history_sync::PUSHNAMES, + &pushname.encode_to_vec(), + ); + emit_len_field( + &mut blob, + tags::history_sync::CONVERSATIONS, + &conv_a.encode_to_vec(), + ); + emit_varint(&mut blob, ((50 << 3) | wire_type::VARINT) as u64); + emit_varint(&mut blob, 7); + emit_len_field(&mut blob, tags::history_sync::NCT_SALT, &[0xAA, 0xBB]); + emit_len_field( + &mut blob, + tags::history_sync::CONVERSATIONS, + &conv_b.encode_to_vec(), ); - assert_eq!(full.conversations_processed, 2); - assert_eq!(full.tc_token_candidates, streamed.tc_token_candidates); + + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&blob).unwrap(); + let compressed = encoder.finish().unwrap(); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + let got_a = stream.next_conversation().unwrap().unwrap(); + let got_b = stream.next_conversation().unwrap().unwrap(); + assert!(stream.next_conversation().unwrap().is_none()); + assert_eq!(got_a, conv_a); + assert_eq!(got_b, conv_b); + + let remainder = stream.remainder().unwrap(); + assert_eq!(remainder.pushnames, vec![pushname]); + assert_eq!(remainder.nct_salt.as_deref(), Some(&[0xAA, 0xBB][..])); + assert!(remainder.conversations.is_empty()); + } + + /// PushName-only / nctSalt-only / empty blobs: zero conversations, complete + /// remainder. + #[test] + fn stream_conversationless_blobs() { + let cases: Vec = vec![ + wa::HistorySync { + sync_type: wa::history_sync::HistorySyncType::PushName as i32, + pushnames: vec![wa::Pushname { + id: Some("5511000000000".into()), + pushname: Some("Me".into()), + }], + ..Default::default() + }, + wa::HistorySync { + sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32, + nct_salt: Some(vec![1, 2, 3]), + ..Default::default() + }, + wa::HistorySync::default(), + ]; + + for hs in cases { + let compressed = encode_and_compress(&hs); + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + assert!(stream.next_conversation().unwrap().is_none()); + let remainder = stream.remainder().unwrap(); + assert_eq!(remainder, hs); + } + } + + /// One corrupt conversation among valid ones: skipped and counted, the + /// stream continues and the remainder stays intact. + #[test] + fn stream_lenient_decode_skips_corrupt_conversation() { + let good = wa::Conversation { + id: "5511111111111@s.whatsapp.net".into(), + ..Default::default() + }; + // Field 1 (id) claims 5 bytes but only 1 follows: prost decode fails. + let corrupt = [0x0A, 0x05, b'x']; + + let mut blob = Vec::new(); + emit_len_field( + &mut blob, + tags::history_sync::CONVERSATIONS, + &good.encode_to_vec(), + ); + emit_len_field(&mut blob, tags::history_sync::CONVERSATIONS, &corrupt); + emit_len_field( + &mut blob, + tags::history_sync::CONVERSATIONS, + &good.encode_to_vec(), + ); + emit_len_field(&mut blob, tags::history_sync::NCT_SALT, &[0x42]); + + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&blob).unwrap(); + let compressed = encoder.finish().unwrap(); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + let mut decoded = Vec::new(); + while let Some(conversation) = stream.next_conversation().unwrap() { + decoded.push(conversation); + } + assert_eq!(decoded, vec![good.clone(), good]); + assert_eq!(stream.skipped_conversations(), 1); + let remainder = stream.remainder().unwrap(); + assert_eq!(remainder.nct_salt.as_deref(), Some(&[0x42][..])); + } + + /// Level 1 still yields the corrupt entry's raw bytes (leniency is a level + /// 2 policy). + #[test] + fn stream_level1_yields_raw_bytes_verbatim() { + let corrupt = [0x0A, 0x05, b'x']; + let mut blob = Vec::new(); + emit_len_field(&mut blob, tags::history_sync::CONVERSATIONS, &corrupt); + + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&blob).unwrap(); + let compressed = encoder.finish().unwrap(); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); assert_eq!( - full.tc_token_candidates.len(), - 1, - "only the DM has a tctoken" + stream.next_conversation_bytes().unwrap(), + Some(&corrupt[..]) + ); + assert!(stream.next_conversation_bytes().unwrap().is_none()); + } + + /// A zero-length conversation entry is yielded (empty slice; decodes to the + /// default Conversation), not conflated with EOF. + #[test] + fn stream_zero_length_conversation() { + let mut blob = Vec::new(); + emit_len_field(&mut blob, tags::history_sync::CONVERSATIONS, &[]); + + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&blob).unwrap(); + let compressed = encoder.finish().unwrap(); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + let conversation = stream.next_conversation().unwrap().unwrap(); + assert_eq!(conversation, wa::Conversation::default()); + assert!(stream.next_conversation().unwrap().is_none()); + } + + /// Truncated zlib stream and truncated length-delimited field both surface + /// clean errors. + #[test] + fn stream_truncated_inputs_error_cleanly() { + let hs = parity_fixture("5511000000000"); + let compressed = encode_and_compress(&hs); + let truncated_zlib = &compressed[..compressed.len() / 2]; + let mut stream = HistorySyncStream::new(truncated_zlib, MAX_DECOMPRESSED); + let mut saw_error = false; + loop { + match stream.next_conversation_bytes() { + Ok(Some(_)) => continue, + Ok(None) => break, + Err(e) => { + saw_error = true; + assert!(matches!( + e, + HistorySyncError::DecompressionError(_) + | HistorySyncError::MalformedProtobuf(_) + )); + break; + } + } + } + assert!(saw_error, "a half zlib stream must not parse cleanly"); + + // A field that claims more payload than the stream carries. + let mut blob = Vec::new(); + emit_varint( + &mut blob, + ((tags::history_sync::CONVERSATIONS << 3) | wire_type::LENGTH_DELIMITED) as u64, ); - assert_eq!(full.msg_secret_records, streamed.msg_secret_records); - assert_eq!(full.msg_secret_records.len(), 1500 + 1); + emit_varint(&mut blob, 100); + blob.extend_from_slice(&[0u8; 10]); + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&blob).unwrap(); + let compressed = encoder.finish().unwrap(); + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + assert!(matches!( + stream.next_conversation_bytes(), + Err(HistorySyncError::MalformedProtobuf(_)) + )); + } + + /// A conversation far larger than the inflate window must force the window + /// to grow and still come out intact. + #[test] + fn stream_window_grows_for_large_conversation() { + let big = wa::Conversation { + id: "5511111111111@s.whatsapp.net".into(), + messages: vec![wa::HistorySyncMsg { + message: Some(wa::WebMessageInfo { + key: wa::MessageKey { + id: Some("BIG".into()), + ..Default::default() + }, + message: Some(wa::Message { + conversation: Some("x".repeat(1_000_000)), + ..Default::default() + }), + ..Default::default() + }), + ..Default::default() + }], + ..Default::default() + }; + let hs = wa::HistorySync { + conversations: vec![big.clone()], + ..Default::default() + }; + let compressed = encode_and_compress(&hs); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + let got = stream.next_conversation().unwrap().unwrap(); + assert_eq!(got, big); + assert!(stream.next_conversation().unwrap().is_none()); + } + + /// Many mid-size conversations force repeated window refills, so field + /// headers straddle inflate chunk boundaries somewhere along the way. + #[test] + fn stream_survives_many_window_refills() { + let mut conversations = Vec::new(); + for i in 0..50u32 { + conversations.push(wa::Conversation { + id: format!("55119{i:08}@s.whatsapp.net"), + messages: vec![wa::HistorySyncMsg { + message: Some(wa::WebMessageInfo { + key: wa::MessageKey { + id: Some(format!("M{i}")), + ..Default::default() + }, + message: Some(wa::Message { + conversation: Some(format!("{i}").repeat(4_000)), + ..Default::default() + }), + ..Default::default() + }), + ..Default::default() + }], + ..Default::default() + }); + } + let hs = wa::HistorySync { + conversations: conversations.clone(), + ..Default::default() + }; + let compressed = encode_and_compress(&hs); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + let mut got = Vec::new(); + while let Some(conversation) = stream.next_conversation().unwrap() { + got.push(conversation); + } + assert_eq!(got, conversations); + } + + /// Inflating past `max_decompressed` must error instead of allocating. + #[test] + fn stream_enforces_decompressed_cap() { + let hs = wa::HistorySync { + conversations: vec![wa::Conversation { + id: "5511111111111@s.whatsapp.net".into(), + messages: vec![wa::HistorySyncMsg { + message: Some(wa::WebMessageInfo { + message: Some(wa::Message { + conversation: Some("y".repeat(64 * 1024)), + ..Default::default() + }), + ..Default::default() + }), + ..Default::default() + }], + ..Default::default() + }], + ..Default::default() + }; + let compressed = encode_and_compress(&hs); + + let mut stream = HistorySyncStream::new(&compressed, 1024); + assert!(matches!( + stream.next_conversation_bytes(), + Err(HistorySyncError::DecompressionError(_)) + )); + } + + /// `remainder()` before exhaustion drains the tail; an unread conversation + /// in it is a loud error, while a conversation-free tail succeeds. + #[test] + fn stream_remainder_before_exhaustion_is_fail_loud() { + let conv = wa::Conversation { + id: "5511111111111@s.whatsapp.net".into(), + ..Default::default() + }; + let hs = wa::HistorySync { + conversations: vec![conv], + nct_salt: Some(vec![9]), + ..Default::default() + }; + let compressed = encode_and_compress(&hs); + let stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + assert!(matches!( + stream.remainder(), + Err(HistorySyncError::UnreadConversations) + )); + + // Without conversations the early call is fine: the drain only meets + // remainder fields. + let hs = wa::HistorySync { + nct_salt: Some(vec![9]), + ..Default::default() + }; + let compressed = encode_and_compress(&hs); + let stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + let remainder = stream.remainder().unwrap(); + assert_eq!(remainder.nct_salt.as_deref(), Some(&[9][..])); + } + + /// Group wire types (3/4) don't exist in HistorySync; the stream mirrors + /// the extractor and rejects them as malformed. + #[test] + fn stream_unknown_wire_type_errors() { + let mut blob = Vec::new(); + emit_varint(&mut blob, ((99 << 3) | wire_type::START_GROUP) as u64); + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&blob).unwrap(); + let compressed = encoder.finish().unwrap(); + + let mut stream = HistorySyncStream::new(&compressed, MAX_DECOMPRESSED); + assert!(matches!( + stream.next_conversation_bytes(), + Err(HistorySyncError::MalformedProtobuf(_)) + )); + } + + /// The extraction pass reports the exact inflated size. + #[test] + fn extraction_reports_exact_decompressed_size() { + let hs = parity_fixture("5511000000000"); + let raw_len = hs.encode_to_vec().len(); + let compressed = encode_and_compress(&hs); + let result = process_history_sync(compressed, None, false).unwrap(); + assert_eq!(result.decompressed_size, raw_len); } } diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 892fa4c5f..9f72142aa 100755 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -6,7 +6,7 @@ use bytes::Bytes; use chrono::{DateTime, Duration, Utc}; use serde::Serialize; use std::fmt; -use std::sync::{Arc, Mutex, OnceLock, RwLock}; +use std::sync::{Arc, OnceLock, RwLock}; use wacore_binary::Node; use wacore_binary::OwnedNodeRef; use wacore_binary::{Jid, MessageId}; @@ -14,24 +14,33 @@ use waproto::whatsapp as wa; /// A lazily-parsed history sync blob. /// -/// Wraps the decompressed protobuf bytes and only decodes on first access. -/// With `Arc` dispatch, all handlers share the same `LazyHistorySync` -/// so `OnceLock` gives parse-once semantics for free. +/// Carries the original **compressed** payload (one immutable `Bytes`, +/// typically ~10x smaller than the inflated form), so holding or queueing the +/// event costs O(compressed) memory. Cheap metadata (`sync_type`, +/// `chunk_order`, `progress`) is available without touching the payload, and +/// with `Arc` dispatch all handlers share the same instance. /// -/// Cheap metadata (`sync_type`, `chunk_order`, `progress`) is available -/// without decoding — useful for filtering events. +/// Three ways at the payload, by increasing cost: +/// - [`stream()`](Self::stream) — conversations one at a time with bounded +/// memory (peak ≈ the largest single conversation), plus a decoded +/// remainder for everything else. +/// - [`decompress()`](Self::decompress) — the raw decompressed protobuf +/// bytes, inflated per call, for custom partial decoding. +/// - [`get()`](Self::get) — full decode, cached; later calls are free. /// -/// Call [`get()`](Self::get) for full access to conversations, pushnames, -/// global settings, past participants, call logs, and everything else in -/// the `wa::HistorySync` proto. +/// A multi-MB chunk takes tens of milliseconds to inflate (plus decode for +/// `get()`). Inside an async handler, prefer doing that work in +/// `spawn_blocking` — clone [`compressed_bytes()`](Self::compressed_bytes) +/// into the closure — when [`decompressed_size()`](Self::decompressed_size) +/// is large. pub struct LazyHistorySync { - /// Decompressed protobuf bytes. Taken (freed) once [`get()`](Self::get) - /// materializes the owned proto, so the two halves don't coexist (~2x the - /// decompressed size) for the event's lifetime. - raw_bytes: Mutex>, - /// Original decompressed size, kept after `raw_bytes` is freed so Debug and - /// [`raw_size()`](Self::raw_size) stay meaningful. - raw_size: usize, + /// Original zlib-compressed payload. Immutable for the event's lifetime: + /// clones are refcount bumps, and every accessor keeps working after + /// [`get()`](Self::get) (no take-dance). + compressed: Bytes, + /// Exact inflated size, counted by the producer's extraction pass; doubles + /// as the inflate cap (a tighter anti-bomb bound than the global ceiling). + decompressed_size: usize, sync_type: i32, chunk_order: Option, progress: Option, @@ -43,39 +52,32 @@ pub struct LazyHistorySync { impl Clone for LazyHistorySync { fn clone(&self) -> Self { - // Common case (not yet decoded): carry the raw bytes for a cheap, lazy - // clone. Once `get()` has freed the raw bytes, carry the decoded proto - // instead (a deep copy, only when cloning an already-inspected blob) so - // the clone stays usable rather than decoding to `None`. - let raw = self.locked_raw().clone(); - let parsed = OnceLock::new(); - if raw.is_none() - && let Some(decoded) = self.parsed.get() - { - let _ = parsed.set(decoded.clone()); - } + // The decode cache is intentionally not carried over: it would deep-copy + // a multi-MB proto. A clone re-inflates on demand from the shared + // compressed bytes. Self { - raw_bytes: Mutex::new(raw), - raw_size: self.raw_size, + compressed: self.compressed.clone(), + decompressed_size: self.decompressed_size, sync_type: self.sync_type, chunk_order: self.chunk_order, progress: self.progress, peer_data_request_session_id: self.peer_data_request_session_id.clone(), - parsed, + parsed: OnceLock::new(), } } } impl LazyHistorySync { pub fn new( - raw_bytes: Bytes, + compressed: Bytes, + decompressed_size: usize, sync_type: i32, chunk_order: Option, progress: Option, ) -> Self { Self { - raw_size: raw_bytes.len(), - raw_bytes: Mutex::new(Some(raw_bytes)), + compressed, + decompressed_size, sync_type, chunk_order, progress, @@ -84,12 +86,6 @@ impl LazyHistorySync { } } - /// Lock the raw-bytes slot, recovering from a poisoned mutex (a poison only - /// means a prior holder panicked; the `Option` is still valid). - fn locked_raw(&self) -> std::sync::MutexGuard<'_, Option> { - self.raw_bytes.lock().unwrap_or_else(|e| e.into_inner()) - } - pub fn with_peer_data_request_session_id(mut self, id: Option) -> Self { self.peer_data_request_session_id = id; self @@ -116,42 +112,54 @@ impl LazyHistorySync { self.peer_data_request_session_id.as_deref() } - /// Full decode of the history sync proto, cached via OnceLock. - /// Returns `None` if decoding fails. - /// - /// On the first successful decode the decompressed `raw_bytes` are freed, so - /// only the owned proto is retained afterwards (not ~2x). A consumer that - /// needs the raw bytes for partial decoding must read [`raw_bytes()`] before - /// calling this; afterwards it returns `None`. - /// - /// [`raw_bytes()`]: Self::raw_bytes - pub fn get(&self) -> Option<&wa::HistorySync> { - let parsed = self.parsed.get_or_init(|| { - // Cheap refcount bump; the lock is released before decoding so a - // concurrent reader isn't blocked by the parse. - let raw = self.locked_raw().clone()?; - waproto::codec::history_sync_decode(&raw[..]) - .ok() - .map(Box::new) - }); - // Free the raw bytes only AFTER the owned proto is committed, so a - // concurrent clone never sees both gone (raw == None implies parsed set). - if parsed.is_some() { - *self.locked_raw() = None; - } - parsed.as_deref() + /// The original zlib-compressed payload. Zero-cost access; `Bytes` clones + /// share the buffer (hand one to `spawn_blocking` for off-runtime + /// consumption). + pub fn compressed_bytes(&self) -> &Bytes { + &self.compressed } - /// The raw decompressed protobuf bytes for custom/partial decoding, or - /// `None` once [`get()`](Self::get) has consumed them on a successful decode. - pub fn raw_bytes(&self) -> Option { - self.locked_raw().clone() + /// Exact size of the decompressed blob in bytes, known without inflating. + pub fn decompressed_size(&self) -> usize { + self.decompressed_size } - /// Size of the decompressed blob in bytes, available even after the raw - /// bytes have been freed by [`get()`](Self::get). - pub fn raw_size(&self) -> usize { - self.raw_size + /// Inflate the payload, returning the raw decompressed protobuf bytes for + /// custom partial decoding. Inflates on EVERY call (no caching) — hold on + /// to the result if it is needed more than once. The exact + /// [`decompressed_size`](Self::decompressed_size) caps the inflate, so a + /// tampered blob fails instead of over-allocating. + pub fn decompress(&self) -> std::io::Result { + wacore_binary::zlib_pool::decompress_zlib_pooled( + &self.compressed, + self.decompressed_size as u64, + ) + .map(Bytes::from) + } + + /// Incremental reader over the payload: conversations one at a time, then + /// everything else as a decoded remainder, without materializing the whole + /// decompressed blob. See [`HistorySyncStream`]. + /// + /// [`HistorySyncStream`]: crate::history_sync::HistorySyncStream + pub fn stream(&self) -> crate::history_sync::HistorySyncStream<'_> { + crate::history_sync::HistorySyncStream::new(&self.compressed, self.decompressed_size as u64) + } + + /// Full decode of the history sync proto, cached via OnceLock: the first + /// call inflates + decodes, later calls are free. Returns `None` if + /// inflating or decoding fails. The compressed payload is kept, so + /// [`decompress()`](Self::decompress) and [`stream()`](Self::stream) keep + /// working afterwards. + pub fn get(&self) -> Option<&wa::HistorySync> { + self.parsed + .get_or_init(|| { + let raw = self.decompress().ok()?; + waproto::codec::history_sync_decode(&raw[..]) + .ok() + .map(Box::new) + }) + .as_deref() } } @@ -165,8 +173,8 @@ impl fmt::Debug for LazyHistorySync { "peer_data_request_session_id", &self.peer_data_request_session_id, ) - .field("raw_size", &self.raw_size) - .field("raw_freed", &self.locked_raw().is_none()) + .field("compressed_size", &self.compressed.len()) + .field("decompressed_size", &self.decompressed_size) .field( "parsed", &self.parsed.get().and_then(|o| o.as_ref()).is_some(), @@ -1242,23 +1250,33 @@ mod tests { use prost::Message; use waproto::whatsapp as wa; - /// Build a HistorySync proto with conversations and encode it. - fn make_history_sync_bytes(conversations: Vec) -> Vec { + /// Build a HistorySync proto with conversations, returning its + /// zlib-compressed wire form plus the exact decompressed size. + fn make_compressed_history_sync(conversations: Vec) -> (Bytes, usize) { + use flate2::{Compression, write::ZlibEncoder}; + use std::io::Write; let hs = wa::HistorySync { sync_type: wa::history_sync::HistorySyncType::InitialBootstrap as i32, conversations, ..Default::default() }; - hs.encode_to_vec() + let raw = hs.encode_to_vec(); + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&raw).unwrap(); + (Bytes::from(encoder.finish().unwrap()), raw.len()) + } + + fn lazy_from(conversations: Vec) -> LazyHistorySync { + let (compressed, raw_len) = make_compressed_history_sync(conversations); + LazyHistorySync::new(compressed, raw_len, 0, None, None) } #[test] fn lazy_history_sync_get_decodes() { - let bytes = make_history_sync_bytes(vec![wa::Conversation { + let lazy = lazy_from(vec![wa::Conversation { id: "chat@s.whatsapp.net".to_string(), ..Default::default() }]); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); let hs = lazy.get().expect("should decode"); assert_eq!(hs.conversations.len(), 1); @@ -1267,11 +1285,10 @@ mod tests { #[test] fn lazy_history_sync_caches_decode() { - let bytes = make_history_sync_bytes(vec![wa::Conversation { + let lazy = lazy_from(vec![wa::Conversation { id: "test@g.us".to_string(), ..Default::default() }]); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); let first = lazy.get().expect("first decode"); let second = lazy.get().expect("second decode"); @@ -1281,22 +1298,24 @@ mod tests { #[test] fn lazy_history_sync_cheap_metadata() { - let bytes = make_history_sync_bytes(vec![]); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 3, Some(2), Some(50)); + let (compressed, raw_len) = make_compressed_history_sync(vec![]); + let lazy = LazyHistorySync::new(compressed.clone(), raw_len, 3, Some(2), Some(50)); assert_eq!(lazy.sync_type(), 3); assert_eq!(lazy.chunk_order(), Some(2)); assert_eq!(lazy.progress(), Some(50)); + assert_eq!(lazy.decompressed_size(), raw_len); + assert_eq!(lazy.compressed_bytes(), &compressed); } #[test] fn lazy_history_sync_peer_data_request_session_id() { - let bytes = make_history_sync_bytes(vec![]); + let (compressed, raw_len) = make_compressed_history_sync(vec![]); - let unset = LazyHistorySync::new(Bytes::from(bytes.clone()), 0, None, None); + let unset = LazyHistorySync::new(compressed.clone(), raw_len, 0, None, None); assert_eq!(unset.peer_data_request_session_id(), None); - let set = LazyHistorySync::new(Bytes::from(bytes), 0, None, None) + let set = LazyHistorySync::new(compressed, raw_len, 0, None, None) .with_peer_data_request_session_id(Some("session-123".to_string())); assert_eq!(set.peer_data_request_session_id(), Some("session-123")); @@ -1306,84 +1325,96 @@ mod tests { } #[test] - fn lazy_history_sync_raw_bytes() { - let bytes = make_history_sync_bytes(vec![wa::Conversation { + fn lazy_history_sync_decompress_yields_raw_proto() { + let lazy = lazy_from(vec![wa::Conversation { id: "raw@s.whatsapp.net".to_string(), ..Default::default() }]); - let raw = bytes.clone(); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); - assert_eq!(lazy.raw_bytes().as_deref(), Some(&raw[..])); - - // Consumer can partial-decode from raw_bytes - let raw_bytes = lazy.raw_bytes().expect("raw still present before get()"); - let decoded = wa::HistorySync::decode(&raw_bytes[..]).expect("should decode"); + // Consumer can partial-decode from the inflated bytes. + let raw = lazy.decompress().expect("inflates"); + assert_eq!(raw.len(), lazy.decompressed_size()); + let decoded = wa::HistorySync::decode(&raw[..]).expect("should decode"); assert_eq!(decoded.conversations[0].id, "raw@s.whatsapp.net"); + + // No caching: a second call inflates again and matches. + assert_eq!(lazy.decompress().expect("inflates again"), raw); } #[test] - fn lazy_history_sync_get_frees_raw_bytes() { - let bytes = make_history_sync_bytes(vec![wa::Conversation { - id: "freed@s.whatsapp.net".to_string(), + fn lazy_history_sync_everything_keeps_working_after_get() { + let lazy = lazy_from(vec![wa::Conversation { + id: "kept@s.whatsapp.net".to_string(), ..Default::default() }]); - let raw_len = bytes.len(); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 7, Some(1), Some(42)); - assert!(lazy.raw_bytes().is_some(), "raw present before get()"); assert_eq!( lazy.get().expect("decodes").conversations[0].id, - "freed@s.whatsapp.net" + "kept@s.whatsapp.net" ); - // The decompressed bytes are released once the owned proto exists. - assert!( - lazy.raw_bytes().is_none(), - "raw freed after a successful get()" - ); - // Metadata survives the free. - assert_eq!(lazy.raw_size(), raw_len); - assert_eq!(lazy.sync_type(), 7); - assert_eq!(lazy.chunk_order(), Some(1)); - assert_eq!(lazy.progress(), Some(42)); - // get() still returns the cached proto after raw is gone. - assert_eq!( - lazy.get().expect("cached").conversations[0].id, - "freed@s.whatsapp.net" - ); + // The compressed payload is kept: decompress() and stream() still work + // after a successful get() (the old take-dance surprise is gone). + let raw = lazy.decompress().expect("decompress after get()"); + assert_eq!(raw.len(), lazy.decompressed_size()); + let mut stream = lazy.stream(); + let conversation = stream + .next_conversation() + .expect("stream after get()") + .expect("one conversation"); + assert_eq!(conversation.id, "kept@s.whatsapp.net"); } #[test] - fn lazy_history_sync_keeps_raw_when_decode_fails() { - // A corrupt blob fails to decode; raw is kept so partial decode / retry - // remains possible (only a successful decode frees it). - let lazy = LazyHistorySync::new(Bytes::from_static(&[0xFF, 0xFF, 0xFF]), 0, None, None); - assert!(lazy.get().is_none()); - assert!( - lazy.raw_bytes().is_some(), - "raw must survive a failed decode" + fn lazy_history_sync_stream_iterates_conversations() { + let lazy = lazy_from(vec![ + wa::Conversation { + id: "first@s.whatsapp.net".to_string(), + ..Default::default() + }, + wa::Conversation { + id: "second@s.whatsapp.net".to_string(), + ..Default::default() + }, + ]); + + let mut stream = lazy.stream(); + assert_eq!( + stream.next_conversation().unwrap().unwrap().id, + "first@s.whatsapp.net" + ); + assert_eq!( + stream.next_conversation().unwrap().unwrap().id, + "second@s.whatsapp.net" + ); + assert!(stream.next_conversation().unwrap().is_none()); + let remainder = stream.remainder().expect("remainder decodes"); + assert!(remainder.conversations.is_empty()); + assert_eq!( + remainder.sync_type(), + wa::history_sync::HistorySyncType::InitialBootstrap ); } #[test] - fn lazy_history_sync_clone_after_get_stays_decodable() { - let bytes = make_history_sync_bytes(vec![wa::Conversation { + fn lazy_history_sync_clone_is_cheap_and_redecodes() { + let lazy = lazy_from(vec![wa::Conversation { id: "cloned@s.whatsapp.net".to_string(), ..Default::default() }]); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); - // Decode on the original, which frees its raw bytes. + // Decode on the original; the clone shares the compressed buffer (no + // deep copy) and re-decodes on demand (the cache isn't carried over). assert_eq!( lazy.get().expect("decodes").conversations[0].id, "cloned@s.whatsapp.net" ); - assert!(lazy.raw_bytes().is_none()); - - // A clone taken AFTER the original decoded must still yield the history - // (the decoded proto is carried over since the raw bytes are gone). let cloned = lazy.clone(); + assert_eq!( + cloned.compressed_bytes().as_ptr(), + lazy.compressed_bytes().as_ptr(), + "clone shares the compressed buffer" + ); assert_eq!( cloned.get().expect("clone still decodes").conversations[0].id, "cloned@s.whatsapp.net" @@ -1391,33 +1422,33 @@ mod tests { } #[test] - fn lazy_history_sync_clone_before_get_is_lazy() { - let bytes = make_history_sync_bytes(vec![wa::Conversation { - id: "lazyclone@s.whatsapp.net".to_string(), - ..Default::default() - }]); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); - - // Cloning before any decode carries the raw bytes (cheap, still lazy). - let cloned = lazy.clone(); - assert!(cloned.raw_bytes().is_some(), "lazy clone carries raw"); - assert_eq!( - cloned.get().expect("decodes").conversations[0].id, - "lazyclone@s.whatsapp.net" - ); + fn lazy_history_sync_empty_proto_decodes_default() { + // A zero-conversation HistorySync still inflates and decodes. + let lazy = lazy_from(vec![]); + let hs = lazy.get().expect("decodes"); + assert!(hs.conversations.is_empty()); } #[test] - fn lazy_history_sync_empty_bytes_decodes_default() { - // Empty protobuf bytes are valid — decode to default HistorySync - let lazy = LazyHistorySync::new(Bytes::new(), 0, None, None); - let hs = lazy.get().expect("empty bytes decode to default"); - assert!(hs.conversations.is_empty()); + fn lazy_history_sync_corrupt_bytes_returns_none() { + // Not a zlib stream: inflating fails, get() yields None, and the + // payload stays available for inspection. + let lazy = LazyHistorySync::new(Bytes::from_static(&[0xFF, 0xFF, 0xFF]), 16, 0, None, None); + assert!(lazy.get().is_none()); + assert!(lazy.decompress().is_err()); + assert_eq!(lazy.compressed_bytes().len(), 3); } #[test] - fn lazy_history_sync_corrupt_bytes_returns_none() { - let lazy = LazyHistorySync::new(Bytes::from_static(&[0xFF, 0xFF, 0xFF]), 0, None, None); + fn lazy_history_sync_undersized_cap_fails_loud() { + // A decompressed_size below the real inflated size trips the inflate + // cap instead of silently over-allocating past the producer's count. + let (compressed, raw_len) = make_compressed_history_sync(vec![wa::Conversation { + id: "capped@s.whatsapp.net".to_string(), + ..Default::default() + }]); + let lazy = LazyHistorySync::new(compressed, raw_len - 1, 0, None, None); + assert!(lazy.decompress().is_err()); assert!(lazy.get().is_none()); } @@ -1437,8 +1468,7 @@ mod tests { }], ..Default::default() }; - let bytes = make_history_sync_bytes(vec![conv]); - let lazy = LazyHistorySync::new(Bytes::from(bytes), 0, None, None); + let lazy = lazy_from(vec![conv]); let hs = lazy.get().expect("should decode"); assert_eq!(hs.conversations[0].messages.len(), 1); From 5f17b29354b80f90e784e2d8714a5c52d7e58c41 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 15:22:43 -0300 Subject: [PATCH 2/9] fix(history-sync): reject zlib streams truncated at a field boundary 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. --- wacore/binary/src/zlib_pool.rs | 22 +++++++++++--- wacore/src/history_sync.rs | 53 ++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 4 deletions(-) diff --git a/wacore/binary/src/zlib_pool.rs b/wacore/binary/src/zlib_pool.rs index 7cd30d5df..78b9a9fbf 100644 --- a/wacore/binary/src/zlib_pool.rs +++ b/wacore/binary/src/zlib_pool.rs @@ -33,6 +33,7 @@ pub struct InflateReader<'a> { total_out: u64, max: u64, eof: bool, + stream_end: bool, } impl<'a> InflateReader<'a> { @@ -60,6 +61,7 @@ impl<'a> InflateReader<'a> { total_out: 0, max, eof: false, + stream_end: false, } } @@ -98,6 +100,13 @@ impl<'a> InflateReader<'a> { self.total_out } + /// Whether zlib reported a proper stream end (terminator + adler32 + /// checksum). An EOF (`ensure` returning false) without this means the + /// input was truncated, not finished. + pub fn stream_ended(&self) -> bool { + self.stream_end + } + fn pump(&mut self) -> io::Result<()> { // Drop the consumed prefix before growing, so the buffer holds roughly // just the record currently being accumulated. @@ -135,11 +144,16 @@ impl<'a> InflateReader<'a> { self.buf.extend_from_slice(&chunk[..produced]); match status { - Status::StreamEnd => self.eof = true, + Status::StreamEnd => { + self.eof = true; + self.stream_end = true; + } // No output produced and not at stream end: distinguish a truncated - // tail (no input left → treat as end) from a stalled/corrupt stream - // (input remains but the decompressor consumed none → error, instead - // of spinning forever since 64 KB of output is always available). + // tail (no input left → treat as end, with `stream_end` left false + // so callers can tell it apart from a real terminator) from a + // stalled/corrupt stream (input remains but the decompressor + // consumed none → error, instead of spinning forever since 64 KB of + // output is always available). // Mirrors the no-progress guard in `decompress_zlib_pooled`. _ if produced == 0 => { if self.in_pos >= self.input.len() { diff --git a/wacore/src/history_sync.rs b/wacore/src/history_sync.rs index 10a848dba..44a608948 100644 --- a/wacore/src/history_sync.rs +++ b/wacore/src/history_sync.rs @@ -132,6 +132,15 @@ impl<'a> FieldWalker<'a> { .ensure(1) .map_err(HistorySyncError::DecompressionError)? { + // Input exhausted at a field boundary is only a clean EOF when zlib + // saw its terminator; otherwise a truncated blob would pass as + // parsed (side effects applied, event dispatched) and the retained + // payload would fail every later get()/decompress(). + if !self.reader.stream_ended() { + return Err(HistorySyncError::MalformedProtobuf( + "zlib stream truncated (missing terminator)".into(), + )); + } return Ok(None); } // A varint is at most 10 bytes (fewer is fine right at EOF). @@ -3261,6 +3270,50 @@ mod tests { assert_eq!(remainder.nct_salt.as_deref(), Some(&[9][..])); } + /// A zlib stream cut exactly at a protobuf field boundary (all fields + /// parse, but no zlib terminator) must NOT pass as a successfully parsed + /// blob: the old retained path rejected it via the full decompress, and a + /// dispatched event would fail every later get()/decompress(). + #[test] + fn truncated_zlib_without_terminator_is_rejected() { + let conv = wa::Conversation { + id: "5511111111111@s.whatsapp.net".into(), + ..Default::default() + }; + let hs = wa::HistorySync { + conversations: vec![conv.clone()], + ..Default::default() + }; + + // Sync-flush makes every written byte inflatable, then drop the + // encoder without finish(): a valid prefix with no terminator, so the + // inflater exhausts input cleanly right at the field boundary. + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&hs.encode_to_vec()).unwrap(); + encoder.flush().unwrap(); + let truncated = encoder.get_ref().clone(); + + // Sanity: the same bytes WITH the terminator parse fine. + let complete = encoder.finish().unwrap(); + assert!(process_history_sync(complete, None, true).is_ok()); + + // Extraction rejects the truncated form outright (no side effects, no + // event). + assert!(matches!( + process_history_sync(truncated.clone(), None, true), + Err(HistorySyncError::MalformedProtobuf(_)) + )); + + // The public stream yields the conversations it can, then surfaces the + // truncation instead of reporting clean EOF. + let mut stream = HistorySyncStream::new(&truncated, MAX_DECOMPRESSED); + assert_eq!(stream.next_conversation().unwrap().unwrap(), conv); + assert!(matches!( + stream.next_conversation(), + Err(HistorySyncError::MalformedProtobuf(_)) + )); + } + /// Group wire types (3/4) don't exist in HistorySync; the stream mirrors /// the extractor and rejects them as malformed. #[test] From 2e1053e5e42a51fc29b777963b6330a400a63eee Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 15:46:34 -0300 Subject: [PATCH 3/9] perf(zlib): inflate directly into the reader window 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). --- wacore/binary/src/zlib_pool.rs | 10 ++++++---- 1 file changed, 6 insertions(+), 4 deletions(-) diff --git a/wacore/binary/src/zlib_pool.rs b/wacore/binary/src/zlib_pool.rs index 78b9a9fbf..8ac2db94f 100644 --- a/wacore/binary/src/zlib_pool.rs +++ b/wacore/binary/src/zlib_pool.rs @@ -115,19 +115,22 @@ impl<'a> InflateReader<'a> { self.cursor = 0; } - let mut chunk = [0u8; Self::CHUNK]; // `decomp` is `Some` for the reader's whole lifetime (only `Drop` takes it), // so this is unreachable in practice; surface it as an error rather than panic. let decomp = self .decomp .as_mut() .ok_or_else(|| io::Error::other("InflateReader used after pool return"))?; + // Inflate straight into the window's spare capacity: a stack chunk + + // extend_from_slice would copy every decompressed byte a second time + // (~10% of a history-sync extraction). + self.buf.reserve(Self::CHUNK); let prev_in = decomp.total_in(); let prev_out = decomp.total_out(); let status = decomp - .decompress( + .decompress_vec( &self.input[self.in_pos..], - &mut chunk, + &mut self.buf, FlushDecompress::None, ) .map_err(|e| io::Error::new(io::ErrorKind::InvalidData, e))?; @@ -141,7 +144,6 @@ impl<'a> InflateReader<'a> { format!("decompressed payload exceeds {} bytes", self.max), )); } - self.buf.extend_from_slice(&chunk[..produced]); match status { Status::StreamEnd => { From f9fbe1724da51b24529d1e27800db630caa84077 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 15:46:34 -0300 Subject: [PATCH 4/9] perf(waproto)!: box the per-message giants out of the repeated-field 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> (construction sites wrap with Box::new; reads auto-deref). --- src/history_sync.rs | 30 +++++----- src/pdo.rs | 2 +- wacore/benches/history_sync_benchmark.rs | 6 +- wacore/src/history_sync.rs | 76 ++++++++++++------------ wacore/src/types/events.rs | 4 +- waproto/build.rs | 9 +++ 6 files changed, 68 insertions(+), 59 deletions(-) diff --git a/src/history_sync.rs b/src/history_sync.rs index e57611b67..131b98f30 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -519,20 +519,20 @@ mod tests { conversations: vec![wa::Conversation { id: chat.to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), id: Some(parent_id.to_string()), participant: None, }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { conversation: Some("historical".to_string()), ..Default::default() - }), + })), message_secret: Some(secret.clone()), ..Default::default() - }), + })), msg_order_id: Some(1), }], ..Default::default() @@ -631,20 +631,20 @@ mod tests { conversations: vec![wa::Conversation { id: chat.to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), id: Some(parent_id.to_string()), participant: None, }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { conversation: Some("bot historical".to_string()), ..Default::default() - }), + })), message_secret: Some(secret.clone()), ..Default::default() - }), + })), msg_order_id: Some(1), }], ..Default::default() @@ -695,18 +695,18 @@ mod tests { } }; wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), id: Some(msg_id.to_string()), participant: None, }, - message: Some(message), + message: Some(Box::new(message)), message_secret: Some(secret.to_vec()), message_timestamp: Some(ts_secs), ..Default::default() - }), + })), msg_order_id: Some(1), } } @@ -985,25 +985,25 @@ mod tests { ..Default::default() }); wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), id: Some(msg_id.to_string()), participant: Some(participant.to_string()), }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { extended_text_message: Some(Box::new(wa::message::ExtendedTextMessage { text: Some("hi".into()), ..Default::default() })), message_context_info, ..Default::default() - }), + })), message_secret: Some(secret.to_vec()), message_timestamp: Some(ts_secs), ..Default::default() - }), + })), msg_order_id: Some(1), } } diff --git a/src/pdo.rs b/src/pdo.rs index a6e886633..58bec1760 100644 --- a/src/pdo.rs +++ b/src/pdo.rs @@ -415,7 +415,7 @@ impl Client { self.core .event_bus .dispatch(wacore::types::events::Event::Message( - Arc::new(message), + Arc::from(message), message_info, )); } diff --git a/wacore/benches/history_sync_benchmark.rs b/wacore/benches/history_sync_benchmark.rs index 2e4e1ea6c..21abf9c76 100644 --- a/wacore/benches/history_sync_benchmark.rs +++ b/wacore/benches/history_sync_benchmark.rs @@ -61,17 +61,17 @@ fn build_realistic_history_sync(n_convos: usize, msgs_per_convo: usize) -> Vec Vec { wa::HistorySyncMsg { - message: Some(web_msg.clone()), + message: Some(Box::new(web_msg.clone())), ..Default::default() } .encode_to_vec() @@ -1802,7 +1802,7 @@ mod tests { id: Some(id.to_string()), ..Default::default() }, - message, + message: message.map(Box::new), message_timestamp: Some(1_700_000_777), ..Default::default() } @@ -2424,7 +2424,7 @@ mod tests { conversations: vec![wa::Conversation { id: chat.to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), @@ -2433,7 +2433,7 @@ mod tests { }, message_secret: Some(Vec::new()), ..Default::default() - }), + })), ..Default::default() }], ..Default::default() @@ -2507,7 +2507,7 @@ mod tests { id: chat.to_string(), messages: vec![ wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), @@ -2516,26 +2516,26 @@ mod tests { }, message_secret: Some(top_level_secret.clone()), ..Default::default() - }), + })), ..Default::default() }, wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(true), id: Some("HIST_CONTEXT".to_string()), participant: None, }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { message_context_info: Some(wa::MessageContextInfo { message_secret: Some(context_secret.clone()), ..Default::default() }), ..Default::default() - }), + })), ..Default::default() - }), + })), ..Default::default() }, ], @@ -2579,7 +2579,7 @@ mod tests { conversations: vec![wa::Conversation { id: chat.to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), @@ -2587,15 +2587,15 @@ mod tests { participant: Some("5511888889999@s.whatsapp.net".to_string()), }, message_secret: Some(top_level_secret.clone()), - message: Some(wa::Message { + message: Some(Box::new(wa::Message { message_context_info: Some(wa::MessageContextInfo { message_secret: Some(context_secret.clone()), ..Default::default() }), ..Default::default() - }), + })), ..Default::default() - }), + })), ..Default::default() }], ..Default::default() @@ -2627,14 +2627,14 @@ mod tests { conversations: vec![wa::Conversation { id: chat.to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), id: Some("HIST_FORWARDED".to_string()), ..Default::default() }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { extended_text_message: Some(Box::new( wa::message::ExtendedTextMessage { text: Some("forwarded".into()), @@ -2650,9 +2650,9 @@ mod tests { ..Default::default() }), ..Default::default() - }), + })), ..Default::default() - }), + })), ..Default::default() }], ..Default::default() @@ -2674,14 +2674,14 @@ mod tests { conversations: vec![wa::Conversation { id: chat.to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(chat.to_string()), from_me: Some(false), id: Some("HIST_NESTED_FORWARDED".to_string()), ..Default::default() }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { view_once_message: Some(Box::new(wa::message::FutureProofMessage { message: Some(Box::new(wa::Message { ephemeral_message: Some(Box::new( @@ -2711,9 +2711,9 @@ mod tests { ..Default::default() }), ..Default::default() - }), + })), ..Default::default() - }), + })), ..Default::default() }], ..Default::default() @@ -2805,7 +2805,7 @@ mod tests { let mut big_msgs = Vec::new(); for i in 0..1500u32 { big_msgs.push(wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(dm.to_string()), from_me: Some(i % 2 == 0), @@ -2815,7 +2815,7 @@ mod tests { message_timestamp: Some(1_700_000_000 + i as u64), message_secret: Some(vec![(i % 251) as u8; 32]), ..Default::default() - }), + })), msg_order_id: Some(i as u64 + 1), }); } @@ -2830,7 +2830,7 @@ mod tests { let group_conv = wa::Conversation { id: group.to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { remote_jid: Some(group.to_string()), from_me: Some(false), @@ -2839,7 +2839,7 @@ mod tests { }, message_secret: Some(vec![0x33u8; 32]), ..Default::default() - }), + })), msg_order_id: Some(1), }], tc_token: Some(vec![0xCDu8; 16]), @@ -3143,17 +3143,17 @@ mod tests { let big = wa::Conversation { id: "5511111111111@s.whatsapp.net".into(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { id: Some("BIG".into()), ..Default::default() }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { conversation: Some("x".repeat(1_000_000)), ..Default::default() - }), + })), ..Default::default() - }), + })), ..Default::default() }], ..Default::default() @@ -3179,17 +3179,17 @@ mod tests { conversations.push(wa::Conversation { id: format!("55119{i:08}@s.whatsapp.net"), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { id: Some(format!("M{i}")), ..Default::default() }, - message: Some(wa::Message { + message: Some(Box::new(wa::Message { conversation: Some(format!("{i}").repeat(4_000)), ..Default::default() - }), + })), ..Default::default() - }), + })), ..Default::default() }], ..Default::default() @@ -3216,13 +3216,13 @@ mod tests { conversations: vec![wa::Conversation { id: "5511111111111@s.whatsapp.net".into(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { - message: Some(wa::Message { + message: Some(Box::new(wa::WebMessageInfo { + message: Some(Box::new(wa::Message { conversation: Some("y".repeat(64 * 1024)), ..Default::default() - }), + })), ..Default::default() - }), + })), ..Default::default() }], ..Default::default() diff --git a/wacore/src/types/events.rs b/wacore/src/types/events.rs index 9f72142aa..2d0d02306 100755 --- a/wacore/src/types/events.rs +++ b/wacore/src/types/events.rs @@ -1457,13 +1457,13 @@ mod tests { let conv = wa::Conversation { id: "chat@s.whatsapp.net".to_string(), messages: vec![wa::HistorySyncMsg { - message: Some(wa::WebMessageInfo { + message: Some(Box::new(wa::WebMessageInfo { key: wa::MessageKey { id: Some("msg-0".to_string()), ..Default::default() }, ..Default::default() - }), + })), msg_order_id: Some(0), }], ..Default::default() diff --git a/waproto/build.rs b/waproto/build.rs index 1bf5ecc06..76ca7d71e 100644 --- a/waproto/build.rs +++ b/waproto/build.rs @@ -57,6 +57,15 @@ fn main() -> std::io::Result<()> { ".whatsapp.SenderKeyStateStructure.SenderSigningKey", ]); + // Box the giant per-message structs out of the repeated-field decode path: + // WebMessageInfo is ~15.6 KB inline (wa::Message ~6.5 KB inside it), so + // prost's push(default) + Vec doubling memcpy'd hundreds of MB per full + // history-sync decode with HistorySyncMsg elements at ~15.7 KB each. + // Boxing collapses the element to pointer size; CodSpeed flamegraphs + // attributed ~94% of a full conversation decode to those copies. + config.boxed(".whatsapp.HistorySyncMsg.message"); + config.boxed(".whatsapp.WebMessageInfo.message"); + // Bytes fields lack serde support; skip them (internal crypto state). config.field_attribute( ".whatsapp.SessionStructure.Chain.ChainKey.key", From 288a18b61abe014cc39ed0dfa5574d4191d6a904 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 16:08:21 -0300 Subject: [PATCH 5/9] fix(history-sync): evaluate handler interest at dispatch time, drop empty-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. --- src/history_sync.rs | 22 ++++++++++++---------- wacore/src/history_sync.rs | 33 +++++++++++++++++++++++++++++---- 2 files changed, 41 insertions(+), 14 deletions(-) diff --git a/src/history_sync.rs b/src/history_sync.rs index 131b98f30..173b4cf8b 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -141,12 +141,10 @@ impl Client { device_snapshot.pn.as_ref().map(|j| j.to_non_ad().user) }; - // Retain (and fully decompress) the blob only when a handler actually - // wants HistorySync. A message-only bot leaves this false, so the - // streaming decompress-and-parse path runs instead of materializing the - // whole payload just to drop it at dispatch. - let retain_history_blob = self.core.event_bus.has_handler_for(EventKind::HistorySync); - + // Always carry the compressed input through (a move, no copy or extra + // inflate); handler interest is evaluated at dispatch time below, so a + // handler that registers while a large blob is being parsed still gets + // the event instead of racing a pre-parse snapshot. // Small blobs (PushName, Recent): decode inline to avoid spawn_blocking overhead. // Large blobs: use blocking thread to avoid stalling the async runtime. const INLINE_THRESHOLD: usize = 256 * 1024; @@ -154,13 +152,12 @@ impl Client { Some(process_history_sync( compressed_data, own_user.as_deref(), - retain_history_blob, + true, )) } else { let (result_tx, result_rx) = futures::channel::oneshot::channel(); let blocking_fut = self.runtime.spawn_blocking(Box::new(move || { - let result = - process_history_sync(compressed_data, own_user.as_deref(), retain_history_blob); + let result = process_history_sync(compressed_data, own_user.as_deref(), true); let _ = result_tx.send(result); })); self.runtime @@ -214,7 +211,12 @@ impl Client { self.store_history_sync_msg_secrets(sync_result.msg_secret_records) .await; - if let Some(compressed) = sync_result.compressed_bytes { + // Dispatch-time interest check: a bot with no HistorySync + // handler drops the compressed payload here (it was a move, + // never a copy), while one registered mid-parse still wins. + if self.core.event_bus.has_handler_for(EventKind::HistorySync) + && let Some(compressed) = sync_result.compressed_bytes + { let lazy_hs = LazyHistorySync::new( compressed, sync_result.decompressed_size, diff --git a/wacore/src/history_sync.rs b/wacore/src/history_sync.rs index 65754a311..35bb36996 100644 --- a/wacore/src/history_sync.rs +++ b/wacore/src/history_sync.rs @@ -1630,15 +1630,17 @@ fn extract_conversation_fields( } } - // tc-token candidate: only for 1:1 chats that actually carry a token. + // tc-token candidate: only for 1:1 chats that actually carry a token. A + // malformed conversation without an id must not emit a candidate either + // (same guard the message-secret extraction applies). + if chat_id.is_empty() || tc_token.is_empty() { + return None; + } if let Some(parts) = wacore_binary::jid::parse_jid_fast(chat_id) && (parts.server == "g.us" || parts.server == "newsletter" || parts.server == "bot") { return None; } - if tc_token.is_empty() { - return None; - } Some(TcTokenCandidate { id: chat_id.to_string(), tc_token: tc_token.to_vec(), @@ -2332,6 +2334,29 @@ mod tests { assert!(result.msg_secret_records.is_empty()); } + /// A (malformed) conversation carrying a tctoken but no id must not emit a + /// candidate under an empty chat id. + #[test] + fn test_tc_token_without_conversation_id_yields_no_candidate() { + let mut conv = Vec::new(); + emit_len_field(&mut conv, tags::conversation::TC_TOKEN, &[0xABu8; 16]); + emit_varint( + &mut conv, + ((tags::conversation::TC_TOKEN_TIMESTAMP << 3) | wire_type::VARINT) as u64, + ); + emit_varint(&mut conv, 1_700_000_123); + let mut hs = Vec::new(); + emit_len_field(&mut hs, tags::history_sync::CONVERSATIONS, &conv); + + let mut encoder = ZlibEncoder::new(Vec::new(), Compression::default()); + encoder.write_all(&hs).unwrap(); + let compressed = encoder.finish().unwrap(); + + let result = process_history_sync(compressed, None, false).unwrap(); + assert!(result.tc_token_candidates.is_empty()); + assert_eq!(result.conversations_processed, 1); + } + /// The presence pre-scan must honor prost merge semantics: a secret carried /// by a LATER occurrence of a repeated message field still yields a record. #[test] From 70801fcbaca7a84191c904a9d12aca0283e21ee1 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 16:27:06 -0300 Subject: [PATCH 6/9] fix(history-sync): let dispatch() own the handler-interest decision MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- src/history_sync.rs | 14 +++++++------- 1 file changed, 7 insertions(+), 7 deletions(-) diff --git a/src/history_sync.rs b/src/history_sync.rs index 173b4cf8b..b5a0af7e4 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -1,4 +1,4 @@ -use crate::types::events::{Event, EventKind, LazyHistorySync}; +use crate::types::events::{Event, LazyHistorySync}; use std::sync::Arc; use wacore::history_sync::{HistoryMsgSecretRecord, TcTokenCandidate, process_history_sync}; use wacore::store::traits::{MsgSecretEntry, TcTokenEntry}; @@ -211,12 +211,12 @@ impl Client { self.store_history_sync_msg_secrets(sync_result.msg_secret_records) .await; - // Dispatch-time interest check: a bot with no HistorySync - // handler drops the compressed payload here (it was a move, - // never a copy), while one registered mid-parse still wins. - if self.core.event_bus.has_handler_for(EventKind::HistorySync) - && let Some(compressed) = sync_result.compressed_bytes - { + // No interest pre-check: dispatch() evaluates handler interest + // against a single bus snapshot (and skips materializing the + // Arc when nobody listens), so deferring to it removes the + // check-to-dispatch race window entirely. Building the event + // is just a Bytes refcount move plus metadata. + if let Some(compressed) = sync_result.compressed_bytes { let lazy_hs = LazyHistorySync::new( compressed, sync_result.decompressed_size, From a835413308b7f9000eafdde0dc7fa3abd78914f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 17:27:58 -0300 Subject: [PATCH 7/9] perf(waproto)!: box statusMentionMessageInfo and messageContextInfo 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. --- src/client/messaging.rs | 4 +- src/features/comments.rs | 4 +- src/features/events.rs | 4 +- src/features/polls.rs | 4 +- src/history_sync.rs | 12 ++-- src/message/tests.rs | 72 ++++++++++++------------ src/send.rs | 4 +- wacore/benches/history_sync_benchmark.rs | 4 +- wacore/src/history_sync.rs | 40 ++++++------- wacore/src/messages.rs | 24 ++++---- wacore/src/msg_secret.rs | 4 +- wacore/src/proto_helpers.rs | 24 ++++---- wacore/src/reporting_token.rs | 12 ++-- waproto/build.rs | 11 ++-- 14 files changed, 113 insertions(+), 110 deletions(-) diff --git a/src/client/messaging.rs b/src/client/messaging.rs index a1b88270e..6d6353ef2 100644 --- a/src/client/messaging.rs +++ b/src/client/messaging.rs @@ -425,10 +425,10 @@ fn build_secret_message_edit( ), remote_key_id: None, }), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(message_secret.to_vec()), ..Default::default() - }), + })), ..Default::default() }) } diff --git a/src/features/comments.rs b/src/features/comments.rs index 75716490f..20c80f9f4 100644 --- a/src/features/comments.rs +++ b/src/features/comments.rs @@ -103,10 +103,10 @@ impl<'a> Comments<'a> { enc_payload: Some(enc_payload), enc_iv: Some(iv.to_vec()), }), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(comment_secret.clone()), ..Default::default() - }), + })), ..Default::default() }; let result = client.send_message(chat, message).await?; diff --git a/src/features/events.rs b/src/features/events.rs index 03d453fbc..4315d4dc1 100644 --- a/src/features/events.rs +++ b/src/features/events.rs @@ -58,10 +58,10 @@ impl<'a> Events<'a> { rand::make_rng::().fill_bytes(&mut secret); secret }; - message.message_context_info = Some(wa::MessageContextInfo { + message.message_context_info = Some(Box::new(wa::MessageContextInfo { message_secret: Some(message_secret.clone()), ..Default::default() - }); + })); let result = self.client.send_message(to, message).await?; Ok((result, message_secret)) diff --git a/src/features/polls.rs b/src/features/polls.rs index 7cc46bdee..5347c2d3e 100644 --- a/src/features/polls.rs +++ b/src/features/polls.rs @@ -89,10 +89,10 @@ impl<'a> Polls<'a> { secret }; - message.message_context_info = Some(wa::MessageContextInfo { + message.message_context_info = Some(Box::new(wa::MessageContextInfo { message_secret: Some(message_secret.clone()), ..Default::default() - }); + })); let result = self.client.send_message(to, message).await?; Ok((result, message_secret)) diff --git a/src/history_sync.rs b/src/history_sync.rs index b5a0af7e4..d9b61628e 100644 --- a/src/history_sync.rs +++ b/src/history_sync.rs @@ -979,12 +979,14 @@ mod tests { ts_secs: u64, bot_prompt: bool, ) -> wa::HistorySyncMsg { - let message_context_info = bot_prompt.then(|| wa::MessageContextInfo { - bot_metadata: Some(wa::BotMetadata { - persona_id: Some("867051314767696".into()), + let message_context_info = bot_prompt.then(|| { + Box::new(wa::MessageContextInfo { + bot_metadata: Some(wa::BotMetadata { + persona_id: Some("867051314767696".into()), + ..Default::default() + }), ..Default::default() - }), - ..Default::default() + }) }); wa::HistorySyncMsg { message: Some(Box::new(wa::WebMessageInfo { diff --git a/src/message/tests.rs b/src/message/tests.rs index 3a53af836..49ee151c2 100644 --- a/src/message/tests.rs +++ b/src/message/tests.rs @@ -4195,7 +4195,7 @@ fn test_is_sender_key_distribution_only() { // SKDM + message_context_info → still true (context_info is metadata) assert!(is_sender_key_distribution_only(&mut wa::Message { sender_key_distribution_message: Some(skdm.clone()), - message_context_info: Some(wa::MessageContextInfo::default()), + message_context_info: Some(Box::default()), ..Default::default() })); @@ -4235,10 +4235,10 @@ fn skdm_only_detection_restores_carrier_fields() { axolotl_sender_key_distribution_message: Some(vec![4, 5, 6]), }, ), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![9, 8, 7]), ..Default::default() - }), + })), ..Default::default() }; @@ -4344,20 +4344,20 @@ fn test_unwrap_device_sent_passthrough() { fn test_unwrap_device_sent_merges_context_info() { let wrapped = wa::Message { // Outer message_context_info (from the DSM envelope) - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![10, 20, 30]), limit_sharing_v2: Some(wa::LimitSharing::default()), ..Default::default() - }), + })), device_sent_message: Some(Box::new(wa::message::DeviceSentMessage { destination_jid: Some("5511999999999@s.whatsapp.net".to_string()), message: Some(Box::new(wa::Message { conversation: Some("hello".to_string()), // Inner has its own message_secret but no limit_sharing_v2 - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![1, 2, 3]), ..Default::default() - }), + })), ..Default::default() })), phash: None, @@ -4383,10 +4383,10 @@ fn test_unwrap_device_sent_merges_context_info() { #[test] fn test_unwrap_device_sent_secret_fallback() { let wrapped = wa::Message { - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![10, 20, 30]), ..Default::default() - }), + })), device_sent_message: Some(Box::new(wa::message::DeviceSentMessage { destination_jid: Some("5511999999999@s.whatsapp.net".to_string()), message: Some(Box::new(wa::Message { @@ -7330,9 +7330,11 @@ fn inner_message_edit(text: &str, next_secret: Option>) -> wa::Message { timestamp_ms: Some(1_770_000_000_000), ..Default::default() })), - message_context_info: next_secret.map(|secret| wa::MessageContextInfo { - message_secret: Some(secret), - ..Default::default() + message_context_info: next_secret.map(|secret| { + Box::new(wa::MessageContextInfo { + message_secret: Some(secret), + ..Default::default() + }) }), ..Default::default() } @@ -8646,10 +8648,10 @@ async fn maybe_capture_inbound_msg_secret_persists_for_bot_chats() { }); let msg = wa::Message { conversation: Some("hi bot".into()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0xAB; 32]), ..Default::default() - }), + })), ..Default::default() }; client.maybe_capture_inbound_msg_secret(&msg, &info).await; @@ -8686,10 +8688,10 @@ async fn maybe_capture_inbound_msg_secret_persists_for_non_bot_chats() { }); let msg = wa::Message { conversation: Some("hi".into()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0xCD; 32]), ..Default::default() - }), + })), ..Default::default() }; client.maybe_capture_inbound_msg_secret(&msg, &info).await; @@ -8743,10 +8745,10 @@ async fn maybe_capture_inbound_msg_secret_persists_for_group_with_bot_mention() })), ..Default::default() })), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0xEE; 32]), ..Default::default() - }), + })), ..Default::default() }; client.maybe_capture_inbound_msg_secret(&msg, &info).await; @@ -8801,10 +8803,10 @@ async fn maybe_capture_inbound_msg_secret_skips_forwarded() { })), ..Default::default() })), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0xFF; 32]), ..Default::default() - }), + })), ..Default::default() }; client.maybe_capture_inbound_msg_secret(&msg, &info).await; @@ -8856,14 +8858,14 @@ async fn maybe_capture_inbound_msg_secret_via_bot_metadata_without_mention() { // No mention at all — just bot_metadata signals the invocation. ..Default::default() })), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0x7B; 32]), bot_metadata: Some(wa::BotMetadata { persona_id: Some("867051314767696".into()), ..Default::default() }), ..Default::default() - }), + })), ..Default::default() }; client.maybe_capture_inbound_msg_secret(&msg, &info).await; @@ -8918,10 +8920,10 @@ async fn bot_only_captures_group_bot_prompt_skips_plain() { }); let plain_msg = wa::Message { conversation: Some("hi".into()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0x01; 32]), ..Default::default() - }), + })), ..Default::default() }; client @@ -8956,14 +8958,14 @@ async fn bot_only_captures_group_bot_prompt_skips_plain() { text: Some("continue".into()), ..Default::default() })), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0x02; 32]), bot_metadata: Some(wa::BotMetadata { persona_id: Some("867051314767696".into()), ..Default::default() }), ..Default::default() - }), + })), ..Default::default() }; client @@ -9009,10 +9011,10 @@ async fn maybe_capture_inbound_msg_secret_keys_under_other_participant() { })), ..Default::default() })), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![0x5A; 32]), ..Default::default() - }), + })), ..Default::default() }; client.maybe_capture_inbound_msg_secret(&msg, &info).await; @@ -9369,10 +9371,10 @@ async fn fanout_capture_lets_subsequent_msmsg_decrypt() { }); let fanout_msg = wa::Message { conversation: Some("hi bot".into()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(secret.to_vec()), ..Default::default() - }), + })), ..Default::default() }; client @@ -9780,7 +9782,7 @@ async fn enc_comment_inbound_dispatches_body_with_parent_link() { })), // Present but secret-less: the outer secret must merge in, not be // dropped because a context already exists. - message_context_info: Some(wa::MessageContextInfo::default()), + message_context_info: Some(Box::default()), ..Default::default() }; let (payload, iv) = wacore::comment::encrypt_comment_with_secret( @@ -9805,10 +9807,10 @@ async fn enc_comment_inbound_dispatches_body_with_parent_link() { }), // WA Web ships the comment's own secret on the OUTER envelope (the // comment msgData), not inside the encrypted body. - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(comment_secret.to_vec()), ..Default::default() - }), + })), ..Default::default() }; let info = Arc::new(MessageInfo { @@ -9912,10 +9914,10 @@ async fn addon_decrypts_right_after_capture_without_flush() { let parent_msg = wa::Message { conversation: Some("hello".to_string()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(secret.to_vec()), ..Default::default() - }), + })), ..Default::default() }; let mk_info = |id: &str| { diff --git a/src/send.rs b/src/send.rs index 1ed2cf1d3..fbaab2794 100644 --- a/src/send.rs +++ b/src/send.rs @@ -1265,10 +1265,10 @@ impl Client { r#type: Some(pin_type as i32), sender_timestamp_ms: Some(wacore::time::now_millis()), }), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_add_on_duration_in_secs: Some(duration_secs), ..Default::default() - }), + })), ..Default::default() }; diff --git a/wacore/benches/history_sync_benchmark.rs b/wacore/benches/history_sync_benchmark.rs index 21abf9c76..68665e1d1 100644 --- a/wacore/benches/history_sync_benchmark.rs +++ b/wacore/benches/history_sync_benchmark.rs @@ -53,10 +53,10 @@ fn build_realistic_history_sync(n_convos: usize, msgs_per_convo: usize) -> Vec wa::Message { 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_ref(), + msg.message_context_info.as_deref(), ); return *inner; } @@ -1075,10 +1075,10 @@ mod device_sent_tests { fn msg_with_secret(secret: &[u8]) -> wa::Message { wa::Message { conversation: Some("hi".into()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(secret.to_vec()), ..Default::default() - }), + })), ..Default::default() } } @@ -1120,10 +1120,10 @@ mod device_sent_tests { #[test] fn wrap_then_unwrap_preserves_non_secret_context_fields() { let inner = wa::Message { - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_add_on_duration_in_secs: Some(604800), ..Default::default() - }), + })), ..Default::default() }; let unwrapped = unwrap_device_sent(wrap_device_sent(inner, "1@s.whatsapp.net".into())); @@ -1217,10 +1217,10 @@ mod device_sent_tests { })), ..Default::default() })), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![1, 2, 3, 4]), ..Default::default() - }), + })), ..Default::default() }, dest, @@ -1244,10 +1244,10 @@ mod device_sent_tests { // mci-only (no content body) assert_splice_matches( wa::Message { - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(vec![7u8; 32]), ..Default::default() - }), + })), ..Default::default() }, dest, @@ -1291,7 +1291,7 @@ mod device_sent_tests { ); let outer_mci = wa::Message { - message_context_info: Some(wa::MessageContextInfo::default()), + message_context_info: Some(Box::default()), ..Default::default() }; assert_eq!( @@ -1332,13 +1332,13 @@ mod device_sent_tests { }, wa::Message { conversation: Some("poll".into()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { // preserved by the merge message_add_on_duration_in_secs: Some(604800), // overwritten by the reporting context message_secret: Some(vec![1u8; 32]), ..Default::default() - }), + })), ..Default::default() }, ] diff --git a/wacore/src/msg_secret.rs b/wacore/src/msg_secret.rs index 0641229a1..f4a1289fd 100644 --- a/wacore/src/msg_secret.rs +++ b/wacore/src/msg_secret.rs @@ -347,10 +347,10 @@ mod tests { assert!(is_bot_context(true, &wa::Message::default())); // bot_metadata on a non-bot (e.g. group) chat is still a bot context. let prompt = wa::Message { - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { bot_metadata: Some(wa::BotMetadata::default()), ..Default::default() - }), + })), ..Default::default() }; assert!(is_bot_context(false, &prompt)); diff --git a/wacore/src/proto_helpers.rs b/wacore/src/proto_helpers.rs index d18087c4d..e5982c331 100644 --- a/wacore/src/proto_helpers.rs +++ b/wacore/src/proto_helpers.rs @@ -642,9 +642,9 @@ pub(crate) fn strip_nested_context_info(msg: &mut wa::Message, always_clear_quot /// - **`thread_id`**: inner if non-empty, otherwise outer /// - **`bot_metadata`**: inner, falling back to outer pub fn merge_dsm_context( - inner: Option, + inner: Option>, outer: Option<&wa::MessageContextInfo>, -) -> Option { +) -> Option> { match (inner, outer) { (None, None) => None, (Some(mut inner), None) => { @@ -654,7 +654,7 @@ pub fn merge_dsm_context( } // Inner was cleared by a WA-Web-style hoist; restore the full context the // sender moved to the outer message, not just the merge subset. - (None, Some(outer)) => Some(outer.clone()), + (None, Some(outer)) => Some(Box::new(outer.clone())), (Some(mut inner), Some(outer)) => { if inner.message_secret.is_none() { inner.message_secret = outer.message_secret.clone(); @@ -1694,7 +1694,7 @@ mod tests { message_secret: Some(vec![1, 2, 3]), ..Default::default() }; - let result = merge_dsm_context(Some(inner.clone()), None).unwrap(); + let result = merge_dsm_context(Some(Box::new(inner.clone())), None).unwrap(); assert_eq!(result.message_secret, Some(vec![1, 2, 3])); } @@ -1741,7 +1741,7 @@ mod tests { message_secret: Some(vec![4, 5, 6]), ..Default::default() }; - let result = merge_dsm_context(Some(inner), Some(&outer)).unwrap(); + let result = merge_dsm_context(Some(Box::new(inner)), Some(&outer)).unwrap(); assert_eq!( result.message_secret, Some(vec![1, 2, 3]), @@ -1759,7 +1759,7 @@ mod tests { message_secret: Some(vec![4, 5, 6]), ..Default::default() }; - let result = merge_dsm_context(Some(inner), Some(&outer)).unwrap(); + let result = merge_dsm_context(Some(Box::new(inner)), Some(&outer)).unwrap(); assert_eq!( result.message_secret, Some(vec![4, 5, 6]), @@ -1783,7 +1783,7 @@ mod tests { limit_sharing_v2: Some(outer_ls), ..Default::default() }; - let result = merge_dsm_context(Some(inner), Some(&outer)).unwrap(); + let result = merge_dsm_context(Some(Box::new(inner)), Some(&outer)).unwrap(); assert_eq!( result.limit_sharing_v2, Some(outer_ls), @@ -1795,7 +1795,7 @@ mod tests { limit_sharing_v2: Some(wa::LimitSharing::default()), ..Default::default() }; - let result = merge_dsm_context(Some(inner_with_ls), None).unwrap(); + let result = merge_dsm_context(Some(Box::new(inner_with_ls)), None).unwrap(); assert_eq!( result.limit_sharing_v2, None, "limit_sharing_v2 should be cleared when outer is None" @@ -1810,7 +1810,7 @@ mod tests { }; // Inner has empty thread_id → should fall back to outer let inner_empty = wa::MessageContextInfo::default(); - let result = merge_dsm_context(Some(inner_empty), Some(&outer)).unwrap(); + let result = merge_dsm_context(Some(Box::new(inner_empty)), Some(&outer)).unwrap(); assert_eq!( result.thread_id.len(), 1, @@ -1822,7 +1822,7 @@ mod tests { thread_id: vec![wa::ThreadId::default(), wa::ThreadId::default()], ..Default::default() }; - let result = merge_dsm_context(Some(inner_filled), Some(&outer)).unwrap(); + let result = merge_dsm_context(Some(Box::new(inner_filled)), Some(&outer)).unwrap(); assert_eq!( result.thread_id.len(), 2, @@ -2031,10 +2031,10 @@ mod tests { url: Some("https://mmg.whatsapp.net/vid".to_string()), ..Default::default() })), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(secret.clone()), ..Default::default() - }), + })), ..Default::default() }; diff --git a/wacore/src/reporting_token.rs b/wacore/src/reporting_token.rs index bd0ca728d..c31d2b47d 100644 --- a/wacore/src/reporting_token.rs +++ b/wacore/src/reporting_token.rs @@ -649,10 +649,10 @@ mod tests { let secret = [0x42u8; MESSAGE_SECRET_SIZE]; let msg = wa::Message { conversation: Some("hi".into()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(secret.to_vec()), ..Default::default() - }), + })), ..Default::default() }; let to: Jid = "5511999999999@s.whatsapp.net".parse().unwrap(); @@ -966,10 +966,10 @@ mod tests { fn test_extract_message_secret() { let secret = vec![0x55u8; MESSAGE_SECRET_SIZE]; let message = wa::Message { - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { message_secret: Some(secret.clone()), ..Default::default() - }), + })), ..Default::default() }; @@ -1378,10 +1378,10 @@ mod tests { // If message already has MessageContextInfo, we should update it, not replace let original = wa::Message { conversation: Some("Test".to_string()), - message_context_info: Some(wa::MessageContextInfo { + message_context_info: Some(Box::new(wa::MessageContextInfo { device_list_metadata_version: Some(42), // Some existing field ..Default::default() - }), + })), ..Default::default() }; diff --git a/waproto/build.rs b/waproto/build.rs index 76ca7d71e..6fed8171c 100644 --- a/waproto/build.rs +++ b/waproto/build.rs @@ -57,14 +57,13 @@ fn main() -> std::io::Result<()> { ".whatsapp.SenderKeyStateStructure.SenderSigningKey", ]); - // Box the giant per-message structs out of the repeated-field decode path: - // WebMessageInfo is ~15.6 KB inline (wa::Message ~6.5 KB inside it), so - // prost's push(default) + Vec doubling memcpy'd hundreds of MB per full - // history-sync decode with HistorySyncMsg elements at ~15.7 KB each. - // Boxing collapses the element to pointer size; CodSpeed flamegraphs - // attributed ~94% of a full conversation decode to those copies. + // Boxed: large (and mostly absent-on-the-wire) submessages whose inline + // form makes prost's repeated-field decode memcpy-bound — every element + // pays push(default) plus Vec-growth copies of the full struct size. config.boxed(".whatsapp.HistorySyncMsg.message"); config.boxed(".whatsapp.WebMessageInfo.message"); + config.boxed(".whatsapp.WebMessageInfo.statusMentionMessageInfo"); + config.boxed(".whatsapp.Message.messageContextInfo"); // Bytes fields lack serde support; skip them (internal crypto state). config.field_attribute( From 8102686face5e0a74b730f3ab0a6938cd7d9a153 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 18:07:15 -0300 Subject: [PATCH 8/9] test(history-sync): deterministic mutation fuzz over the stream and extractor 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. --- wacore/src/history_sync.rs | 62 ++++++++++++++++++++++++++++++++++++++ 1 file changed, 62 insertions(+) diff --git a/wacore/src/history_sync.rs b/wacore/src/history_sync.rs index 6bc43968d..1c67819f5 100644 --- a/wacore/src/history_sync.rs +++ b/wacore/src/history_sync.rs @@ -3339,6 +3339,68 @@ mod tests { )); } + /// Deterministic mutation fuzz: every byte-level corruption of a valid + /// blob must surface as a clean error or lenient skip, never a panic, on + /// both the public stream and the extraction pass. + #[test] + fn stream_and_extractor_survive_mutated_inputs() { + let hs = wa::HistorySync { + conversations: vec![ + wa::Conversation { + id: "5511111111111@s.whatsapp.net".into(), + ..Default::default() + }, + wa::Conversation { + id: "5511222222222@s.whatsapp.net".into(), + ..Default::default() + }, + ], + pushnames: vec![wa::Pushname { + id: Some("5511000000000".into()), + pushname: Some("Me".into()), + }], + nct_salt: Some(vec![1, 2, 3, 4]), + ..Default::default() + }; + let compressed = encode_and_compress(&hs); + + let mut seed = 0x9E37_79B9u32; + let mut next = move || { + seed ^= seed << 13; + seed ^= seed >> 17; + seed ^= seed << 5; + seed + }; + + for _ in 0..10_000 { + let mut mutated = compressed.clone(); + for _ in 0..=(next() % 3) { + match next() % 4 { + 0 if !mutated.is_empty() => { + let len = mutated.len(); + mutated.truncate(next() as usize % len); + } + _ if !mutated.is_empty() => { + let len = mutated.len(); + let idx = next() as usize % len; + mutated[idx] ^= (next() % 255 + 1) as u8; + } + _ => {} + } + } + + let mut stream = HistorySyncStream::new(&mutated, MAX_DECOMPRESSED); + loop { + match stream.next_conversation() { + Ok(Some(_)) => continue, + Ok(None) | Err(_) => break, + } + } + let _ = stream.remainder(); + let _ = process_history_sync(mutated, None, true); + } + } + /// Group wire types (3/4) don't exist in HistorySync; the stream mirrors /// the extractor and rejects them as malformed. #[test] From 1623c5f0f64a4a9168b89abc00c71427f2082006 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Jo=C3=A3o=20Lucas?= Date: Thu, 11 Jun 2026 18:08:47 -0300 Subject: [PATCH 9/9] test(history-sync): appease clippy while-let in the fuzz loop --- wacore/src/history_sync.rs | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/wacore/src/history_sync.rs b/wacore/src/history_sync.rs index 1c67819f5..ed14ecfc9 100644 --- a/wacore/src/history_sync.rs +++ b/wacore/src/history_sync.rs @@ -3390,12 +3390,7 @@ mod tests { } let mut stream = HistorySyncStream::new(&mutated, MAX_DECOMPRESSED); - loop { - match stream.next_conversation() { - Ok(Some(_)) => continue, - Ok(None) | Err(_) => break, - } - } + while let Ok(Some(_)) = stream.next_conversation() {} let _ = stream.remainder(); let _ = process_history_sync(mutated, None, true); }