perf(receipt): aggregate offline delivery receipts per chat like WA Web

src/client.rs

@@ -474,6 +474,11 @@ pub struct Client {
     pub(crate) offline_sync_notifier: Arc<event_listener::Event>,
     /// Flag indicating offline sync has completed (received ib offline stanza).
     pub(crate) offline_sync_completed: 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.
+    pub(crate) offline_receipt_buffer:
+        std::sync::Mutex<Vec<Arc<crate::types::message::MessageInfo>>>,
     /// Number of history sync tasks currently queued or running.
     pub(crate) history_sync_tasks_in_flight: Arc<AtomicUsize>,
     /// Notifier triggered when history sync work becomes idle.

src/client/lifecycle.rs

@@ -201,6 +201,7 @@ impl Client {
             prekey_upload_lock: Arc::new(async_lock::Mutex::new(())),
             offline_sync_notifier: Arc::new(event_listener::Event::new()),
             offline_sync_completed: Arc::new(AtomicBool::new(false)),
+            offline_receipt_buffer: std::sync::Mutex::new(Vec::new()),
             history_sync_tasks_in_flight: Arc::new(AtomicUsize::new(0)),
             history_sync_idle_notifier: Arc::new(event_listener::Event::new()),
             outbound_flush: Arc::new(crate::flush_scope::FlushScope::new()),
@@ -378,6 +379,7 @@ impl Client {
         self.is_ready.store(false, Ordering::Relaxed);
         self.is_connected.store(false, Ordering::Relaxed);
         self.offline_sync_completed.store(false, Ordering::Relaxed);
+        self.clear_offline_receipt_buffer();
         self.offline_batch.reset();
         self.outbound_flush.reopen();
 
@@ -488,6 +490,15 @@ impl Client {
         self.is_running.store(false, Ordering::Relaxed);
         self.shutdown_notifier.notify();
 
+        // Drain buffered offline receipts into the flush window before
+        // closing it, so a disconnect mid-offline-sync still acks the
+        // already-processed backlog (issue #571 semantics). close() only stops
+        // outbound task spawns, not buffering, so a message still in flight can
+        // re-buffer after this drain; those entries are dropped by the
+        // connection-state reset (clear_offline_receipt_buffer) and the server
+        // redelivers their messages on the next connect, where they are
+        // re-acked fresh.
+        self.flush_offline_receipts();
         // Prevent late receipt producers from escaping the drain window.
         self.outbound_flush.close();
         self.outbound_flush
@@ -537,6 +548,7 @@ impl Client {
         self.auto_reconnect_errors
             .store(Self::RECONNECT_BACKOFF_STEP, Ordering::Relaxed);
 
+        self.flush_offline_receipts();
         self.outbound_flush.close();
         self.outbound_flush
             .flush(&*self.runtime, std::time::Duration::from_secs(2))
@@ -561,6 +573,7 @@ impl Client {
         info!("Reconnecting immediately (expected disconnect).");
         self.expected_disconnect.store(true, Ordering::Relaxed);
 
+        self.flush_offline_receipts();
         self.outbound_flush.close();
         self.outbound_flush
             .flush(&*self.runtime, std::time::Duration::from_secs(2))
@@ -641,6 +654,7 @@ impl Client {
         self.pending_device_sync.clear().await;
         // Reset offline sync state for next connection
         self.offline_sync_completed.store(false, Ordering::Relaxed);
+        self.clear_offline_receipt_buffer();
         self.offline_batch.reset();
         self.offline_sync_metrics
             .active

src/client/sessions.rs

@@ -42,6 +42,11 @@ impl Client {
             // Old workers holding the previous semaphore Arc will finish normally.
             self.swap_message_semaphore(64);
 
+            // The flag flip above happens-before this drain takes the buffer
+            // lock, so late offline receipts either land in this flush or
+            // observe the flag and send 1:1 (see try_buffer_offline_receipt).
+            self.flush_offline_receipts();
+
             self.offline_sync_notifier.notify(usize::MAX);
 
             self.core

src/message/dispatch.rs

@@ -61,7 +61,15 @@ impl Client {
     }
 
     /// Spawn a delivery receipt, tracked so `disconnect()` can flush it (issue #571).
+    ///
+    /// Offline-drained messages are buffered instead and flushed as aggregate
+    /// `<receipt>` stanzas when the offline sync completes, collapsing a
+    /// reconnect backlog of N receipts into ~1 stanza per (chat, author)
+    /// (WA Web `sendAggregateOfflineReceipts`). Live messages stay 1:1.
     fn spawn_delivery_receipt(self: &Arc<Self>, info: &Arc<MessageInfo>) {
+        if info.is_offline && self.try_buffer_offline_receipt(info) {
+            return;
+        }
         let client = self.clone();
         let info = Arc::clone(info);
         self.outbound_flush.spawn(&*self.runtime, async move {