feat(recv)!: batch the inbound commit pipeline during the offline drain

agent_docs/e2e_testing.md

@@ -38,7 +38,7 @@ Use `wait_for_event()` with predicates instead of arbitrary sleeps. This is both
 ```rust
 // GOOD: event-driven — returns as soon as the event arrives
 let event = client_b
-    .wait_for_event(15, |e| matches!(e, Event::Message(msg, _) if msg.conversation.as_deref() == Some("hello")))
+    .wait_for_event(15, |e| e.messages().any(|m| m.message.conversation.as_deref() == Some("hello")))
     .await?;
 
 // BAD: arbitrary sleep — wastes time or causes flaky failures
@@ -60,7 +60,7 @@ tokio::time::sleep(Duration::from_millis(100)).await;
 client_a.client.send_message(jid_b.clone(), message).await?;
 
 // Client reconnects automatically and receives from offline queue
-let event = client_b.wait_for_event(30, |e| matches!(e, Event::Message(..))).await?;
+let event = client_b.wait_for_event(30, |e| matches!(e, Event::Messages(_))).await?;
 ```
 
 For full disconnects (no auto-reconnect):

examples/benchmark.rs

@@ -79,7 +79,7 @@ fn main() {
         let bot = builder
             .on_event_for(
                 &[
-                    EventKind::Message,
+                    EventKind::Messages,
                     EventKind::PairingQrCode,
                     EventKind::Connected,
                     EventKind::LoggedOut,
@@ -88,12 +88,12 @@ fn main() {
                     let admin_scan_url = admin_scan_url.clone();
                     async move {
                         match &*event {
-                            Event::Message(msg, info) => {
-                                if let Some(text) = msg.text_content()
-                                    && text == "ping"
-                                {
-                                    let ctx =
-                                        MessageContext::from_arc(Arc::clone(msg), info, client);
+                            Event::Messages(batch) => {
+                                for m in batch {
+                                    if m.message.text_content() != Some("ping") {
+                                        continue;
+                                    }
+                                    let ctx = MessageContext::from_inbound(m, Arc::clone(&client));
                                     info!("Received text ping, sending pong...");
 
                                     let pong_text = format!("pong {}", ctx.info.id);

examples/durability_hook.rs

@@ -95,57 +95,75 @@ impl InboxArchiver {
 
 #[async_trait::async_trait]
 impl InboundDurabilityHook for InboxArchiver {
-    async fn on_message(
+    async fn on_messages(
         &self,
         _client: Arc<Client>,
-        info: &MessageInfo,
-        message: &wa::Message,
+        batch: &[whatsapp_rust::types::events::InboundMessage],
     ) -> anyhow::Result<()> {
-        let key: CommitKey = (
-            info.source.chat.to_string(),
-            info.source.sender.to_string(),
-            info.id.clone(),
-        );
-
-        // Idempotency: a redelivery (or a replay after a crash between commit and
-        // ack) can hand us the same key more than once. Check, but only record it
-        // as committed AFTER the durable write below succeeds.
-        if self
-            .seen
-            .lock()
-            .map_err(|_| anyhow::anyhow!("seen lock poisoned"))?
-            .contains(&key)
+        // Live traffic arrives one message at a time; an offline drain hands
+        // over a whole batch. Either way the commit below is a single append +
+        // fsync, so the durability cost amortizes over the batch.
+        let mut lines = String::new();
+        let mut keys: Vec<CommitKey> = Vec::with_capacity(batch.len());
         {
-            info!("[{}] already committed, skipping (dedup)", info.id);
-            return Ok(());
+            // Idempotency: a redelivery (or a replay after a crash between
+            // commit and ack) can hand us the same keys more than once. Check,
+            // but only record them as committed AFTER the durable write below
+            // succeeds.
+            let seen = self
+                .seen
+                .lock()
+                .map_err(|_| anyhow::anyhow!("seen lock poisoned"))?;
+            for m in batch {
+                let key: CommitKey = (
+                    m.info.source.chat.to_string(),
+                    m.info.source.sender.to_string(),
+                    m.info.id.clone(),
+                );
+                // Dedup against the archive AND earlier entries of this same
+                // batch, so one fsync can never append a key twice.
+                if seen.contains(&key) || keys.contains(&key) {
+                    info!("[{}] already committed, skipping (dedup)", m.info.id);
+                    continue;
+                }
+                // Sanitize so the tab-delimited archive stays parseable on restart.
+                let preview = m
+                    .message
+                    .conversation
+                    .as_deref()
+                    .unwrap_or("<non-text>")
+                    .replace(['\t', '\n'], " ");
+                lines.push_str(&format!("{}\t{}\t{}\t{preview}\n", key.0, key.1, key.2));
+                keys.push(key);
+            }
         }
 
-        // Sanitize so the tab-delimited archive stays parseable on restart.
-        let preview = message
-            .conversation
-            .as_deref()
-            .unwrap_or("<non-text>")
-            .replace(['\t', '\n'], " ");
-        let line = format!("{}\t{}\t{}\t{preview}\n", key.0, key.1, key.2);
-
-        // Durable commit on a blocking thread: append then fsync. Returning Ok
-        // only after sync_all means "safe to ack"; any error returns Err, so the
-        // ack is suppressed and the server redelivers the message later. The hook
-        // is awaited on the receive path, so the disk I/O goes to spawn_blocking.
-        let file = Arc::clone(&self.file);
-        tokio::task::spawn_blocking(move || -> std::io::Result<()> {
-            let mut file = file.lock().expect("file lock poisoned");
-            file.write_all(line.as_bytes())?;
-            file.sync_all()
-        })
-        .await
-        .map_err(|e| anyhow::anyhow!("archive write task failed: {e}"))??;
-
-        self.seen
-            .lock()
-            .map_err(|_| anyhow::anyhow!("seen lock poisoned"))?
-            .insert(key);
-        info!("[{}] committed durably: {preview}", info.id);
+        if !keys.is_empty() {
+            // Durable commit on a blocking thread: append then fsync — all-or-
+            // nothing for the batch. Returning Ok only after sync_all means
+            // "safe to ack every message"; any error returns Err, so the acks
+            // are suppressed and the server redelivers the batch later. The
+            // hook is awaited on the receive path, so disk I/O goes to
+            // spawn_blocking.
+            let file = Arc::clone(&self.file);
+            tokio::task::spawn_blocking(move || -> std::io::Result<()> {
+                let mut file = file.lock().expect("file lock poisoned");
+                file.write_all(lines.as_bytes())?;
+                file.sync_all()
+            })
+            .await
+            .map_err(|e| anyhow::anyhow!("archive write task failed: {e}"))??;
+
+            let mut seen = self
+                .seen
+                .lock()
+                .map_err(|_| anyhow::anyhow!("seen lock poisoned"))?;
+            let count = keys.len();
+            for key in keys {
+                seen.insert(key);
+            }
+            info!("committed {count} message(s) durably in one fsync");
+        }
         Ok(())
     }
 }

src/bot.rs

@@ -143,7 +143,7 @@ pub struct MessageContext {
 
 impl MessageContext {
     /// Builds a context from borrowed parts, deep-cloning `message`. Prefer
-    /// [`MessageContext::from_arc`]/[`MessageContext::from_event`] when an
+    /// [`MessageContext::from_arc`]/[`MessageContext::from_inbound`] when an
     /// `Arc<wa::Message>` is already at hand (the event bus always has one).
     pub fn from_parts(message: &wa::Message, info: &MessageInfo, client: Arc<Client>) -> Self {
         Self::from_arc(Arc::new(message.clone()), info, client)
@@ -157,9 +157,11 @@ impl MessageContext {
         }
     }
 
-    pub fn from_event(event: &Event, client: Arc<Client>) -> Option<Self> {
-        let (msg, info) = event.as_message()?;
-        Some(Self::from_arc(Arc::clone(msg), info, client))
+    pub fn from_inbound(
+        inbound: &wacore::types::events::InboundMessage,
+        client: Arc<Client>,
+    ) -> Self {
+        Self::from_arc(Arc::clone(&inbound.message), &inbound.info, client)
     }
 
     #[cfg_attr(feature = "tracing", tracing::instrument(name = "wa.bot.send_message", level = "debug", skip_all, fields(chat = %self.info.source.chat.observe()), err(Debug)))]
@@ -671,16 +673,34 @@ impl<B, T, H, R> BotBuilder<B, T, H, R> {
 
     /// Run `handler` for every incoming message, with a ready
     /// [`MessageContext`] (reply/react/edit helpers included).
+    ///
+    /// [`Event::Messages`] batches (one per commit during an offline drain,
+    /// single-message on live traffic) are fanned out here in arrival order,
+    /// awaiting each handler before the next — per-message bots keep their
+    /// ergonomics and gain in-batch ordering.
+    ///
+    /// The handler is CALLED for every message in the batch up front and the
+    /// returned futures then run in order (an `async` closure runs no body
+    /// code at call time, so for the typical handler this is unobservable).
+    /// Interleaving call+await instead would hold a `MessageContext` across
+    /// an await, which is not `Send` on wasm32.
     pub fn on_message<F, Fut>(self, handler: F) -> Self
     where
         F: Fn(MessageContext) -> Fut + Send + Sync + 'static,
         Fut: Future<Output = ()> + Send + 'static,
     {
-        self.on_event_for(&[EventKind::Message], move |event, client| {
-            let fut = MessageContext::from_event(&event, client).map(&handler);
+        self.on_event_for(&[EventKind::Messages], move |event, client| {
+            // Futures are built before the async block: `MessageContext` is
+            // not `Send` on wasm32 (the Client's trait objects aren't), so it
+            // must never be held across an await — only the handler futures
+            // (which the `Fut: Send` bound covers) may cross one.
+            let futures: Vec<Fut> = event
+                .messages()
+                .map(|m| handler(MessageContext::from_inbound(m, Arc::clone(&client))))
+                .collect();
             async move {
-                if let Some(fut) = fut {
-                    fut.await
+                for future in futures {
+                    future.await;
                 }
             }
         })
@@ -1495,7 +1515,7 @@ mod tests {
         let handlers = vec![
             RegisteredHandler {
                 callback: noop.clone(),
-                interest: EventInterest::of(&[EventKind::Message]),
+                interest: EventInterest::of(&[EventKind::Messages]),
             },
             RegisteredHandler {
                 callback: noop,
@@ -1504,7 +1524,7 @@ mod tests {
         ];
 
         let interest = combined_interest(&handlers);
-        assert!(interest.wants(EventKind::Message));
+        assert!(interest.wants(EventKind::Messages));
         assert!(interest.wants(EventKind::PairingQrCode));
         assert!(!interest.wants(EventKind::Receipt));
     }

src/client.rs

@@ -358,6 +358,9 @@ pub struct Client {
     /// Write-behind buffer for inbound messageSecret captures; readers check
     /// it before the backend so the durable write can leave the receive lane.
     pub(crate) msg_secret_buffer: Arc<crate::msg_secret_buffer::MsgSecretWriteBuffer>,
+    /// Accumulates decrypted messages during the offline drain for per-batch
+    /// commit (WA Web MessageProcessorCache parity).
+    pub(crate) inbound_commit_batch: crate::message::commit_batch::InboundCommitBatcher,
     pub(crate) media_conn: Arc<RwLock<Option<crate::mediaconn::MediaConn>>>,
 
     pub(crate) is_logged_in: Arc<AtomicBool>,
@@ -522,7 +525,13 @@ pub struct Client {
     /// WhatsApp Web waits for this before sending passive tasks (prekey upload, active IQ, presence).
     pub(crate) offline_sync_notifier: Arc<event_listener::Event>,
     /// Flag indicating offline sync has completed (received ib offline stanza).
+    /// Flips only AFTER the drain-tail commit, so the tail's acks still join
+    /// the aggregate offline-receipt drain.
     pub(crate) offline_sync_completed: Arc<AtomicBool>,
+    /// Once-guard for the drain finisher (the semaphore swap is not
+    /// idempotent). Separate from `offline_sync_completed` because the finish
+    /// runs off the read loop and the flag must flip only after its commit.
+    pub(crate) offline_sync_finish_started: Arc<AtomicBool>,
     /// Delivery receipts buffered during offline sync, flushed as aggregate
     /// `<receipt>` stanzas at completion (WA Web `sendAggregateOfflineReceipts`).
     /// Empty (zero capacity) outside the offline window.

src/client/adapters.rs

@@ -85,13 +85,80 @@ impl Client {
     }
 
     /// [`flush_signal_cache`](Self::flush_signal_cache) with error logging instead of propagation.
+    ///
+    /// Both of these are safe only when the caller holds the message
+    /// processing permit or the batcher is known inactive: they persist the
+    /// WHOLE cache, including ratchet advances of drain entries that may not
+    /// have a durable buffered row yet. Everything else must go through the
+    /// `_batch_safe` variants below.
     pub(crate) async fn flush_signal_cache_logged(&self, context: &str, id: Option<&str>) {
         if let Err(e) = self.flush_signal_cache().await {
-            if let Some(id) = id {
-                log::error!("Failed to flush signal cache ({context} {id}): {e:?}");
-            } else {
-                log::error!("Failed to flush signal cache ({context}): {e:?}");
+            log_signal_flush_error(context, id, &e);
+        }
+    }
+
+    /// Signal-cache flush that is safe while the offline drain is active.
+    ///
+    /// During the drain, decrypted messages accumulate in the commit batcher
+    /// with no durable buffered copy; flushing the cache from an unrelated
+    /// path (a retry receipt, a send, an identity change) would persist their
+    /// ratchet advances, and a crash/teardown that then drops the entries
+    /// turns each redelivery into an ackable duplicate — silent loss for hook
+    /// consumers. So in drain mode this routes through the batcher: commit
+    /// the pending entries (rows first) and flush under the processing
+    /// permit. Outside the drain it is exactly [`Self::flush_signal_cache`].
+    ///
+    /// Must NOT be called while holding the processing permit (it acquires
+    /// it); permit-holding paths commit via the batcher directly.
+    pub(crate) async fn flush_signal_cache_batch_safe(&self) -> Result<(), anyhow::Error> {
+        if self.inbound_commit_batch.is_active() {
+            if let Some(client) = self.self_weak.get().and_then(|w| w.upgrade()) {
+                if !client
+                    .flush_inbound_commits_under_permit(false, None, None)
+                    .await
+                {
+                    return Err(anyhow::anyhow!(
+                        "inbound drain batch commit failed; Signal cache left unflushed so the server redelivers"
+                    ));
+                }
+                // The is_active() check above races the drain finisher: if the
+                // batcher deactivated while we waited for the permit, the commit
+                // took an empty batch in live mode and reported durable WITHOUT
+                // flushing (it had no drain rows to tie an advance to). We are
+                // now live with no uncommitted drain entries, so fall through to
+                // the raw flush to persist our out-of-band advance — the same
+                // path a caller that found the batcher already inactive takes.
+                // While still draining, the commit already flushed under the
+                // permit, so return.
+                if self.inbound_commit_batch.is_active() {
+                    return Ok(());
+                }
+            } else if self.inbound_commit_batch.has_entries() {
+                return Err(anyhow::anyhow!(
+                    "client dropping with uncommitted drain entries; skipping Signal flush"
+                ));
             }
         }
+        self.flush_signal_cache().await
+    }
+
+    /// [`flush_signal_cache_batch_safe`](Self::flush_signal_cache_batch_safe)
+    /// with error logging instead of propagation.
+    pub(crate) async fn flush_signal_cache_batch_safe_logged(
+        &self,
+        context: &str,
+        id: Option<&str>,
+    ) {
+        if let Err(e) = self.flush_signal_cache_batch_safe().await {
+            log_signal_flush_error(context, id, &e);
+        }
+    }
+}
+
+fn log_signal_flush_error(context: &str, id: Option<&str>, e: &anyhow::Error) {
+    if let Some(id) = id {
+        log::error!("Failed to flush signal cache ({context} {id}): {e:?}");
+    } else {
+        log::error!("Failed to flush signal cache ({context}): {e:?}");
     }
 }

src/client/device_registry.rs

@@ -673,7 +673,7 @@ impl Client {
                 }
             }
         }
-        self.flush_signal_cache_logged("delete_sessions_for_devices", None)
+        self.flush_signal_cache_batch_safe_logged("delete_sessions_for_devices", None)
             .await;
     }