perf(history-sync)!: store the compressed payload and expose a streaming reader

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()
@@ -147,31 +141,23 @@ 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;
         let parse_result = if compressed_data.len() < INLINE_THRESHOLD {
             Some(process_history_sync(
                 compressed_data,
                 own_user.as_deref(),
-                retain_history_blob,
-                compressed_size_hint,
+                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,
-                    compressed_size_hint,
-                );
+                let result = process_history_sync(compressed_data, own_user.as_deref(), true);
                 let _ = result_tx.send(result);
             }));
             self.runtime
@@ -225,9 +211,15 @@ impl Client {
                 self.store_history_sync_msg_secrets(sync_result.msg_secret_records)
                     .await;
 
-                if let Some(decompressed) = sync_result.decompressed_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(
-                        decompressed,
+                        compressed,
+                        sync_result.decompressed_size,
                         notification.sync_type().into(),
                         notification.chunk_order,
                         notification.progress,
@@ -529,20 +521,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()
@@ -570,6 +562,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 =
@@ -590,20 +633,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()
@@ -654,18 +697,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),
         }
     }
@@ -944,25 +987,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),
         }
     }

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,
             ));
     }

wacore/benches/history_sync_benchmark.rs

@@ -61,17 +61,17 @@ fn build_realistic_history_sync(n_convos: usize, msgs_per_convo: usize) -> Vec<u
                 }
             };
             messages.push(wa::HistorySyncMsg {
-                message: Some(wa::WebMessageInfo {
+                message: Some(Box::new(wa::WebMessageInfo {
                     key: wa::MessageKey {
                         remote_jid: Some(chat.clone()),
                         from_me: Some(from_me),
                         id: Some(format!("MSGID{c:04}{m:04}ABCDEF")),
                         participant: None,
                     },
-                    message: Some(inner),
+                    message: Some(Box::new(inner)),
                     message_timestamp: Some(1_700_000_000 + (c * msgs_per_convo + m) as u64),
                     ..Default::default()
-                }),
+                })),
                 ..Default::default()
             });
         }
@@ -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()))
+        });
+}

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,
         }
     }
 
@@ -92,6 +94,19 @@ 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
+    }
+
+    /// 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.
@@ -100,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))?;
@@ -126,14 +144,18 @@ impl<'a> InflateReader<'a> {
                 format!("decompressed payload exceeds {} bytes", self.max),
             ));
         }
-        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() {
@@ -225,8 +247,12 @@ pub fn decompress_zlib_pooled(compressed: &[u8], max_size: u64) -> io::Result<Ve
         // every multi-MB history-sync chunk. 2x the compressed length is a
         // conservative first guess (zlib here compresses ~2-5x): it rarely
         // overshoots the real size, so it cuts reallocations without inflating
-        // peak memory. Bounded by `cap` so a bad guess can't exceed the limit.
-        let estimated = compressed.len().saturating_mul(2).clamp(4096, cap);
+        // peak memory. Bounded by `cap` so a bad guess can't exceed the limit;
+        // the floor also bows to `cap` because callers now pass exact (possibly
+        // tiny) decompressed sizes as the limit, where a fixed 4096 floor would
+        // invert the clamp and panic.
+        let floor = 4096.min(cap);
+        let estimated = compressed.len().saturating_mul(2).clamp(floor, cap);
         if scratch.capacity() < estimated {
             scratch.reserve(estimated - scratch.capacity());
         }